mpmd: Refactor device initialization for better parallelizability

Note: This doesn't add any concurrency, rather, it changes the
structure of the code to allow that. Notable changes:
- All prop tree inits in one place
- No access to containers in methods that might be run in parallel
- Split initialization and claiming in mpmd_mboard_impl, calling ctor
  will no longer run the full initialization.
- Added comments to identify parallelizable spots

3 files changed, 125 insertions, 92 deletions

diff --git a/host/lib/usrp/mpmd/mpmd_impl.cpp b/host/lib/usrp/mpmd/mpmd_impl.cpp
index 3afa6dfc1..cb5dfa68d 100644
--- a/host/lib/usrp/mpmd/mpmd_impl.cpp
+++ b/host/lib/usrp/mpmd/mpmd_impl.cpp
@@ -85,16 +85,30 @@ namespace {
     }
 
 
+    /*! Initialize property tree for a single device.
+     *
+     * \param tree Property tree reference (to the whole tree)
+     * \param mb_path Subtree path for this device
+     * \param mb Reference to the actual device
+     */
     void init_property_tree(
             uhd::property_tree::sptr tree,
             fs_path mb_path,
             mpmd_mboard_impl *mb
     ) {
-        if (not tree->exists(fs_path("/name"))) {
+        /*** Device info ****************************************************/
+        if (not tree->exists("/name")) {
             tree->create<std::string>("/name")
                 .set(mb->device_info.get("name", "Unknown MPM device"))
             ;
         }
+        tree->create<std::string>(mb_path / "name")
+            .set(mb->device_info.get("type", "UNKNOWN"));
+        tree->create<std::string>(mb_path / "serial")
+            .set(mb->device_info.get("serial", "n/a"));
+        tree->create<std::string>(mb_path / "connection")
+            .set(mb->device_info.get("connection", "UNKNOWN"));
+        tree->create<size_t>(mb_path / "link_max_rate").set(1e9 / 8);
 
         /*** Clocking *******************************************************/
         tree->create<std::string>(mb_path / "clock_source/value")
@@ -328,31 +342,59 @@ mpmd_impl::mpmd_impl(const device_addr_t& device_args)
     : usrp::device3_impl()
     , _device_args(device_args)
 {
+    const device_addrs_t mb_args = separate_device_addr(device_args);
+    const size_t num_mboards = mb_args.size();
+    _mb.reserve(num_mboards);
+    const bool serialize_init = device_args.has_key("serialize_init");
+    const bool skip_init = device_args.has_key("skip_init");
     UHD_LOGGER_INFO("MPMD")
-        << "Initializing device with args: " << device_args.to_string();
+        << "Initializing " << num_mboards << " device(s) "
+        << (serialize_init ? "serially " : "in parallel ")
+        << "with args: " << device_args.to_string();
+
+    // First, claim all the devices (so we own them and no one else can claim
+    // them).
+    // This can be parallelized as long as uptrs are stored in the right spot;
+    // they need to be correctly indexed.
+    for (size_t mb_i = 0; mb_i < num_mboards; ++mb_i) {
+        UHD_LOG_DEBUG("MPMD", "Claiming mboard " << mb_i);
+        _mb.push_back(claim_and_make(mb_args[mb_i]));
+    }
 
-    const device_addrs_t mb_args = separate_device_addr(device_args);
-    _mb.reserve(mb_args.size());
+    // Next figure out the number of base xport addresses. This way, we
+    // can run _mb[*]->init() in parallel on all the _mb.
+    // This can *not* be parallelized.
+    std::vector<size_t> base_xport_addr(num_mboards, 2); // Starts at 2 [sic]
+    for (size_t mb_i = 0; mb_i < num_mboards-1; ++mb_i) {
+        base_xport_addr[mb_i+1] = base_xport_addr[mb_i] + _mb[mb_i]->num_xbars;
+    }
 
-    // This can theoretically be parallelized, but then we want to make sure
-    // we're distributing crossbar local addresses in some orderly fashion.
-    // At the very least, _xbar_local_addr_ctr needs to become atomic.
-    for (size_t mb_i = 0; mb_i < mb_args.size(); ++mb_i) {
-        _mb.push_back(setup_mb(mb_i, mb_args[mb_i]));
+    if (not skip_init) {
+        // Run the actual device initialization. This can run in parallel.
+        for (size_t mb_i = 0; mb_i < num_mboards; ++mb_i) {
+            // Note: This is the only place we do compat number checks. They're
+            // effectively disabled for skip_init=1
+            setup_mb(_mb[mb_i].get(), mb_i, base_xport_addr[mb_i]);
+        }
+    } else {
+        UHD_LOG_DEBUG("MPMD", "Claimed device, but skipped init.");
     }
 
     // Init the prop tree before the blocks get set up -- they might need access
     // to some of the properties. This also means that the prop tree is pristine
     // at this point in time.
+    // This might be parallelized, need to verify the prop tree can handle the
+    // concurrent accesses. Would shave of milliseconds per device -- probably
+    // not worth it.
     for (size_t mb_i = 0; mb_i < mb_args.size(); ++mb_i) {
         init_property_tree(_tree, fs_path("/mboards") / mb_i, _mb[mb_i].get());
     }
 
-    if (not device_args.has_key("skip_init")) {
-        // This might be parallelized. std::tasks would probably be a good way to
-        // do that if we want to.
+    if (not skip_init) {
+        // This can be parallelized, because the blocks of individual mboards
+        // live on different subtrees.
         for (size_t mb_i = 0; mb_i < mb_args.size(); ++mb_i) {
-            setup_rfnoc_blocks(mb_i, mb_args[mb_i]);
+            setup_rfnoc_blocks(_mb[mb_i].get(), mb_i, mb_args[mb_i]);
         }
 
         // FIXME this section only makes sense for when the time source is external.
@@ -364,10 +406,10 @@ mpmd_impl::mpmd_impl(const device_addr_t& device_args)
         }
 
         auto filtered_block_args = device_args; // TODO actually filter
-        // Blocks will finalize their own setup in this function. They have (and
-        // might need) full access to the prop tree, the timekeepers, etc.
-        setup_rpc_blocks(filtered_block_args,
-                device_args.has_key("serialize_init"));
+        // Blocks will finalize their own setup in this function. They have
+        // (and might need) full access to the prop tree, the timekeepers, etc.
+        // This is already internally parallelized.
+        setup_rpc_blocks(filtered_block_args, serialize_init);
     } else {
         UHD_LOG_INFO("MPMD", "Claimed device without full initialization.");
     }
@@ -381,14 +423,12 @@ mpmd_impl::~mpmd_impl()
 /*****************************************************************************
  * Private methods
  ****************************************************************************/
-mpmd_mboard_impl::uptr mpmd_impl::setup_mb(
-    const size_t mb_index,
+mpmd_mboard_impl::uptr mpmd_impl::claim_and_make(
     const uhd::device_addr_t& device_args
 ) {
     const std::string rpc_addr = device_args.get(xport::MGMT_ADDR_KEY);
     UHD_LOGGER_DEBUG("MPMD")
-        << "Initializing mboard " << mb_index
-        << ". Device args: `" << device_args.to_string()
+        << "Device args: `" << device_args.to_string()
         << "'. RPC address: " << rpc_addr
     ;
 
@@ -398,13 +438,21 @@ mpmd_mboard_impl::uptr mpmd_impl::setup_mb(
             << device_args.to_string());
         throw uhd::runtime_error("Could not determine device RPC address.");
     }
-    auto mb = mpmd_mboard_impl::make(device_args, rpc_addr);
+    return mpmd_mboard_impl::make(device_args, rpc_addr);
+}
+
+void mpmd_impl::setup_mb(
+    mpmd_mboard_impl *mb,
+    const size_t mb_index,
+    const size_t base_xport_addr
+) {
 
     // Check the compatibility number
     UHD_LOGGER_TRACE("MPMD") << boost::format(
             "Checking MPM compat number against ours: %i.%i")
             % MPM_COMPAT_NUM[0] % MPM_COMPAT_NUM[1];
-    const auto compat_num = mb->rpc->request<std::vector<size_t>>("get_mpm_compat_num");
+    const auto compat_num =
+        mb->rpc->request<std::vector<size_t>>("get_mpm_compat_num");
     UHD_LOGGER_TRACE("MPMD") << boost::format(
             "Compat number received: %d.%d")
             % compat_num[0] % compat_num[1];
@@ -425,34 +473,19 @@ mpmd_mboard_impl::uptr mpmd_impl::setup_mb(
         throw uhd::runtime_error("MPM compatibility number mismatch.");
     }
 
-    if (device_args.has_key("skip_init")) {
-        return mb;
-    }
-
+    UHD_LOG_DEBUG("MPMD", "Initializing mboard " << mb_index);
+    mb->init();
     for (size_t xbar_index = 0; xbar_index < mb->num_xbars; xbar_index++) {
-        mb->set_xbar_local_addr(xbar_index, allocate_xbar_local_addr());
+        mb->set_xbar_local_addr(xbar_index, base_xport_addr + xbar_index);
     }
-
-    const fs_path mb_path = fs_path("/mboards") / mb_index;
-    _tree->create<std::string>(mb_path / "name")
-        .set(mb->device_info.get("type", "UNKNOWN"));
-    _tree->create<std::string>(mb_path / "serial")
-        .set(mb->device_info.get("serial", "n/a"));
-    _tree->create<std::string>(mb_path / "connection")
-        .set(mb->device_info.get("connection", "remote"));
-
-    _tree->create<size_t>(mb_path / "link_max_rate").set(1e9 / 8);
-
-    return mb;
 }
 
 void mpmd_impl::setup_rfnoc_blocks(
+    mpmd_mboard_impl* mb,
     const size_t mb_index,
     const uhd::device_addr_t& ctrl_xport_args
 ) {
-    auto &mb = _mb[mb_index];
-    mb->num_xbars = mb->rpc->request<size_t>("get_num_xbars");
-    UHD_LOG_TRACE("MPM",
+    UHD_LOG_TRACE("MPMD",
         "Mboard " << mb_index << " reports " << mb->num_xbars << " crossbar(s)."
     );
 
@@ -523,17 +556,6 @@ void mpmd_impl::setup_rpc_blocks(
     }
 }
 
-size_t mpmd_impl::allocate_xbar_local_addr()
-{
-    const size_t new_local_addr = _xbar_local_addr_ctr++;
-    if (new_local_addr > MPMD_CROSSBAR_MAX_LADDR) {
-        throw uhd::runtime_error("Too many crossbars.");
-    }
-
-    return new_local_addr;
-}
-
-
 /*****************************************************************************
  * Find, Factory & Registry
  ****************************************************************************/
diff --git a/host/lib/usrp/mpmd/mpmd_impl.hpp b/host/lib/usrp/mpmd/mpmd_impl.hpp
index 6d7b93b10..e71e83af6 100644
--- a/host/lib/usrp/mpmd/mpmd_impl.hpp
+++ b/host/lib/usrp/mpmd/mpmd_impl.hpp
@@ -30,7 +30,10 @@ class mpmd_mboard_impl
     using dev_info = std::map<std::string, std::string>;
 
     /*** Structors ***********************************************************/
-    /*!
+    /*! Ctor: Claim device or throw an exception on failure.
+     *
+     * Does not initialize the device.
+     *
      * \param mb_args Device args that pertain to this motherboard
      * \param ip_addr RPC client will attempt to connect to this IP address
      */
@@ -50,6 +53,9 @@ class mpmd_mboard_impl
             const std::string& addr
     );
 
+    /*** Init ****************************************************************/
+    void init();
+
     /*** Public attributes ***************************************************/
     //! These are the args given by the user, with some filtering/preprocessing
     uhd::device_addr_t mb_args;
@@ -62,9 +68,6 @@ class mpmd_mboard_impl
     // to be populated at all.
     std::vector<uhd::device_addr_t> dboard_info;
 
-    //! Number of RFNoC crossbars on this device
-    size_t num_xbars = 0;
-
     /*! Reference to the RPC client for this motherboard
      *
      * We store a shared ptr, because we might share it with some of the RFNoC
@@ -72,6 +75,8 @@ class mpmd_mboard_impl
      */
     uhd::rpc_client::sptr rpc;
 
+    //! Number of RFNoC crossbars on this device
+    const size_t num_xbars;
 
     /*************************************************************************
      * API
@@ -188,20 +193,35 @@ public:
     /*************************************************************************
      * Private methods/helpers
      ************************************************************************/
+    /*! Claim a device and create a reference to the mpmd_mboard_impl object.
+     *
+     * Does not initialize the device (see setup_mb() for that).
+     */
+    mpmd_mboard_impl::uptr claim_and_make(
+        const uhd::device_addr_t& dev_args
+    );
+
     /*! Initialize a single motherboard
      *
-     * - See mpmd_mboard_impl ctor for details
-     * - Also allocates the local crossbar addresses
+     * This is where mpmd_mboard_impl::init() is called.
+     * Also assigns the local crossbar addresses.
+     *
+     * \param mb Reference to the mboard class
+     * \param mb_index Index number of the mboard that's being initialized
+     * \param device_args Device args
+     *
      */
-    mpmd_mboard_impl::uptr setup_mb(
-            const size_t mb_i,
-            const uhd::device_addr_t& dev_addr
+    void setup_mb(
+        mpmd_mboard_impl *mb,
+        const size_t mb_index,
+        const size_t base_xport_addr
     );
 
     //! Setup all RFNoC blocks running on mboard \p mb_i
     void setup_rfnoc_blocks(
-            const size_t mb_i,
-            const uhd::device_addr_t& block_args
+        mpmd_mboard_impl* mb,
+        const size_t mb_i,
+        const uhd::device_addr_t& block_args
     );
 
     //! Configure all blocks that require access to an RPC client
@@ -210,13 +230,6 @@ public:
         const bool serialize_init
     );
 
-    /*! Returns a valid local address for a crossbar
-     *
-     * \returns Valid local address
-     * \throws uhd::runtime_error if there are no more local addresses
-     */
-    size_t allocate_xbar_local_addr();
-
     /*! Return the index of the motherboard given the local address of a
      * crossbar
      */
@@ -229,10 +242,6 @@ public:
     uhd::device_addr_t _device_args;
     //! Stores a list of mboard references
     std::vector<mpmd_mboard_impl::uptr> _mb;
-
-    //! A counter for distributing local addresses to crossbars
-    // No-one touches this except allocate_xbar_local_addr(), gotcha?
-    size_t  _xbar_local_addr_ctr = 2;
 };
 
 }} /* namespace uhd::mpmd */
diff --git a/host/lib/usrp/mpmd/mpmd_mboard_impl.cpp b/host/lib/usrp/mpmd/mpmd_mboard_impl.cpp
index 0ca996d28..59983414a 100644
--- a/host/lib/usrp/mpmd/mpmd_mboard_impl.cpp
+++ b/host/lib/usrp/mpmd/mpmd_mboard_impl.cpp
@@ -111,12 +111,16 @@ mpmd_mboard_impl::mpmd_mboard_impl(
                   mpmd_impl::MPM_RPC_PORT
               ),
               mpmd_impl::MPM_RPC_GET_LAST_ERROR_CMD))
+  , num_xbars(rpc->request<size_t>("get_num_xbars"))
+  // xbar_local_addrs is not yet valid after this!
+  , xbar_local_addrs(num_xbars, 0xFF)
   , _xport_mgr(xport::mpmd_xport_mgr::make(mb_args))
 {
     UHD_LOGGER_TRACE("MPMD")
         << "Initializing mboard, connecting to RPC server address: "
         << rpc_server_addr
         << " mboard args: " << mb_args.to_string()
+        << " number of crossbars: " << num_xbars
     ;
 
     _claimer_task = claim_device_and_make_task(rpc, mb_args);
@@ -124,22 +128,16 @@ mpmd_mboard_impl::mpmd_mboard_impl(
         measure_rpc_latency(rpc, MPMD_MEAS_LATENCY_DURATION);
     }
 
-    // No one else can now claim the device.
-    if (mb_args_.has_key("skip_init")) {
-        UHD_LOG_DEBUG("MPMD", "Claimed device, but skipped init.");
-        return;
-    }
-
-    init_device(rpc, mb_args);
-    // RFNoC block clocks are now on. Noc-IDs can be read back.
-
-    auto device_info_dict = rpc->request<dev_info>("get_device_info");
+    /// Get device info
+    const auto device_info_dict = rpc->request<dev_info>("get_device_info");
     for (const auto &info_pair : device_info_dict) {
         device_info[info_pair.first] = info_pair.second;
     }
     UHD_LOGGER_TRACE("MPMD")
         << "MPM reports device info: " << device_info.to_string();
-    auto dboards_info = rpc->request<std::vector<dev_info>>("get_dboard_info");
+    /// Get dboard info
+    const auto dboards_info =
+        rpc->request<std::vector<dev_info>>("get_dboard_info");
     UHD_ASSERT_THROW(this->dboard_info.size() == 0);
     for (const auto &dboard_info_dict : dboards_info) {
         uhd::device_addr_t this_db_info;
@@ -151,11 +149,6 @@ mpmd_mboard_impl::mpmd_mboard_impl(
             << ": " << this_db_info.to_string();
         this->dboard_info.push_back(this_db_info);
     }
-
-    // Initialize properties
-    this->num_xbars = rpc->request<size_t>("get_num_xbars");
-    // xbar_local_addrs is not yet valid after this!
-    this->xbar_local_addrs.resize(this->num_xbars, 0xFF);
 }
 
 mpmd_mboard_impl::~mpmd_mboard_impl()
@@ -168,6 +161,15 @@ mpmd_mboard_impl::~mpmd_mboard_impl()
 }
 
 /*****************************************************************************
+ * Init
+ ****************************************************************************/
+void mpmd_mboard_impl::init()
+{
+    init_device(rpc, mb_args);
+    // RFNoC block clocks are now on. Noc-IDs can be read back.
+}
+
+/*****************************************************************************
  * API
  ****************************************************************************/
 void mpmd_mboard_impl::set_xbar_local_addr(