mirror of
https://github.com/EttusResearch/uhd.git
synced 2026-10-25 17:36:10 +00:00
1da439c020
UHD hosts <4.10 speak mprpc and unconditionally connect to the RPC port (49601) of any device they discover. When they reach a gRPC-based MPM device, the protocol mismatch crashes uhd_find_device. UHd host 4.10, prevents this by using rpc_version key from discovery response and skips RPC attempt if mprpc device found during uhd_find_devices. This change moves the RPC port to 49602. Now, old UHD <4.10 hosts can still discover both mprpc and grpc devices. But their mprpc connection attempt on 49601 during uhd_find_devices is refused by grpc device (running on new rpc port 49602) and is marked as not reachable instead of crashing.
375 lines
16 KiB
C++
375 lines
16 KiB
C++
//
|
|
// Copyright 2017 Ettus Research, a National Instruments Company
|
|
//
|
|
// SPDX-License-Identifier: GPL-3.0-or-later
|
|
//
|
|
|
|
#include "mpmd_impl.hpp"
|
|
#include <uhd/exception.hpp>
|
|
#include <uhd/types/component_file.hpp>
|
|
#include <uhd/utils/cast.hpp>
|
|
#include <uhd/utils/static.hpp>
|
|
#include <uhd/utils/tasks.hpp>
|
|
#include <uhdlib/utils/prefs.hpp>
|
|
#include <boost/format.hpp>
|
|
#include <algorithm>
|
|
#include <chrono>
|
|
#include <exception>
|
|
#include <future>
|
|
#include <memory>
|
|
#include <mutex>
|
|
#include <random>
|
|
#include <string>
|
|
#include <thread>
|
|
#include <vector>
|
|
|
|
using namespace uhd;
|
|
using namespace uhd::mpmd;
|
|
|
|
namespace {
|
|
/*************************************************************************
|
|
* Local constants
|
|
************************************************************************/
|
|
//! Most pessimistic time for a CHDR query to go to device and back
|
|
const double MPMD_CHDR_MAX_RTT = 0.02;
|
|
//! MPM Compatibility number {MAJOR, MINOR}
|
|
const std::vector<size_t> MPM_COMPAT_NUM = {7, 0};
|
|
/*! Default number of motherboards to initialize in parallel at once.
|
|
*
|
|
* This caps the number of threads spawned during parallel initialization. It
|
|
* can be overridden via the 'init_threads' device arg to provide a new cap.
|
|
*/
|
|
const size_t MPMD_DEFAULT_INIT_THREADS = 1;
|
|
|
|
/*************************************************************************
|
|
* Helper functions
|
|
************************************************************************/
|
|
void reset_time_synchronized(uhd::property_tree::sptr tree)
|
|
{
|
|
const size_t n_mboards = tree->list("/mboards").size();
|
|
UHD_LOGGER_DEBUG("MPMD") << "Synchronizing " << n_mboards << " timekeepers...";
|
|
auto get_time_last_pps = [tree]() {
|
|
return tree->access<time_spec_t>(fs_path("/mboards/0/time/pps")).get();
|
|
};
|
|
auto end_time = std::chrono::steady_clock::now() + std::chrono::milliseconds(1100);
|
|
auto time_last_pps = get_time_last_pps();
|
|
UHD_LOG_DEBUG("MPMD", "Waiting for PPS clock edge...");
|
|
while (time_last_pps == get_time_last_pps()) {
|
|
if (std::chrono::steady_clock::now() > end_time) {
|
|
throw uhd::runtime_error("Board 0 may not be getting a PPS signal!\n"
|
|
"No PPS detected within the time interval.\n"
|
|
"See the application notes for your device.\n");
|
|
}
|
|
std::this_thread::sleep_for(std::chrono::milliseconds(10));
|
|
}
|
|
UHD_LOG_DEBUG("MPMD", "Setting all timekeepers to 0...");
|
|
for (size_t mboard_idx = 0; mboard_idx < n_mboards; mboard_idx++) {
|
|
tree->access<time_spec_t>(fs_path("/mboards") / mboard_idx / "time" / "pps")
|
|
.set(time_spec_t(0.0));
|
|
}
|
|
|
|
UHD_LOG_DEBUG("MPMD", "Waiting for next PPS edge...");
|
|
std::this_thread::sleep_for(std::chrono::seconds(1));
|
|
|
|
UHD_LOG_DEBUG("MPMD", "Verifying all timekeepers are aligned...");
|
|
auto get_time_now = [tree](const size_t mb_index) {
|
|
return tree->access<time_spec_t>(fs_path("/mboards") / mb_index / "time/now")
|
|
.get();
|
|
};
|
|
for (size_t m = 1; m < n_mboards; m++) {
|
|
time_spec_t time_0 = get_time_now(0);
|
|
time_spec_t time_i = get_time_now(m);
|
|
if (time_i < time_0 or (time_i - time_0) > time_spec_t(MPMD_CHDR_MAX_RTT)) {
|
|
UHD_LOGGER_WARNING("MULTI_USRP")
|
|
<< boost::format("Detected time deviation between board %d and board 0.\n"
|
|
"Board 0 time is %f seconds.\n"
|
|
"Board %d time is %f seconds.\n")
|
|
% m % time_0.get_real_secs() % m % time_i.get_real_secs();
|
|
}
|
|
}
|
|
}
|
|
|
|
/*! Throw an exception if compat numbers don't match.
|
|
*
|
|
* \param component Name of the component for which we're checking the
|
|
* compat number (for logging and exceptions strings).
|
|
* \param expected Tuple of 2 integers representing MAJOR.MINOR compat
|
|
* number.
|
|
* \param actual Tuple of 2 integers representing MAJOR.MINOR compat
|
|
* number.
|
|
* \param advice_on_failure A string that is appended to the error message
|
|
* when compat number mismatches have occurred.
|
|
*/
|
|
void assert_compat_number_throw(const std::string& component,
|
|
const std::vector<size_t>& expected,
|
|
const std::vector<size_t>& actual,
|
|
const std::string& advice_on_failure = "")
|
|
{
|
|
UHD_ASSERT_THROW(expected.size() == 2);
|
|
UHD_ASSERT_THROW(actual.size() == 2);
|
|
UHD_LOGGER_TRACE("MPMD") << "Checking " << component
|
|
<< " compat number. Expected: " << expected[0] << "."
|
|
<< expected[1] << " Actual: " << actual[0] << "."
|
|
<< actual[1];
|
|
|
|
if (actual[0] != expected[0]) {
|
|
const std::string err_msg =
|
|
str(boost::format("%s major compat number mismatch. "
|
|
"Expected: %i.%i Actual: %i.%i.%s%s")
|
|
% component % expected[0] % expected[1] % actual[0] % actual[1]
|
|
% (advice_on_failure.empty() ? "" : " ") % advice_on_failure);
|
|
UHD_LOG_ERROR("MPMD", err_msg);
|
|
throw uhd::runtime_error(err_msg);
|
|
}
|
|
if (actual[1] < expected[1]) {
|
|
const std::string err_msg =
|
|
str(boost::format("%s minor compat number mismatch. "
|
|
"Expected: %i.%i Actual: %i.%i.%s%s")
|
|
% component % expected[0] % expected[1] % actual[0] % actual[1]
|
|
% (advice_on_failure.empty() ? "" : " ") % advice_on_failure);
|
|
UHD_LOG_ERROR("MPMD", err_msg);
|
|
throw uhd::runtime_error(err_msg);
|
|
}
|
|
if (actual[1] > expected[1]) {
|
|
const std::string err_msg =
|
|
str(boost::format("%s minor compat number mismatch. "
|
|
"Expected: %i.%i Actual: %i.%i")
|
|
% component % expected[0] % expected[1] % actual[0] % actual[1]);
|
|
UHD_LOG_WARNING("MPMD", err_msg);
|
|
}
|
|
}
|
|
} // namespace
|
|
|
|
/*****************************************************************************
|
|
* Static class attributes
|
|
****************************************************************************/
|
|
const std::string mpmd_impl::MPM_FINDALL_KEY = "find_all";
|
|
const size_t mpmd_impl::MPM_DISCOVERY_PORT = 49600;
|
|
const std::string mpmd_impl::MPM_DISCOVERY_PORT_KEY = "discovery_port";
|
|
const size_t mpmd_impl::MPM_RPC_PORT = 49602;
|
|
const std::string mpmd_impl::MPM_RPC_PORT_KEY = "rpc_port";
|
|
const std::string mpmd_impl::MPM_DISCOVERY_CMD = "MPM-DISC";
|
|
const std::string mpmd_impl::MPM_ECHO_CMD = "MPM-ECHO";
|
|
|
|
/*****************************************************************************
|
|
* Structors
|
|
****************************************************************************/
|
|
mpmd_impl::mpmd_impl(const device_addr_t& device_args)
|
|
: rfnoc_device(), _device_args(device_args)
|
|
{
|
|
const device_addrs_t mb_args_without_prefs = separate_device_addr(device_args);
|
|
device_addrs_t mb_args;
|
|
for (size_t i = 0; i < mb_args_without_prefs.size(); ++i) {
|
|
mb_args.push_back(prefs::get_usrp_args(mb_args_without_prefs[i]));
|
|
UHD_ASSERT_THROW(mb_args.back().has_key(RPC_VERSION_KEY));
|
|
if (mb_args.back().get(RPC_VERSION_KEY) != RPC_VERSION) {
|
|
UHD_LOG_THROW(uhd::runtime_error,
|
|
"MPMD",
|
|
"RPC version mismatch on device "
|
|
<< mb_args.back().to_string() << ". Expected: " << RPC_VERSION
|
|
<< " Actual: " << device_args.get(RPC_VERSION_KEY)
|
|
<< ". Please align the version of MPM on your USRP device to the "
|
|
"host driver version.");
|
|
}
|
|
}
|
|
const size_t num_mboards = mb_args.size();
|
|
_mb.resize(num_mboards);
|
|
// Maximum number of motherboards to initialize in parallel at once. This
|
|
// can be overridden via the 'init_threads' device arg, and defaults to
|
|
// MPMD_DEFAULT_INIT_THREADS.
|
|
const size_t max_init_threads = std::max<size_t>(
|
|
1, device_args.cast<size_t>("init_threads", MPMD_DEFAULT_INIT_THREADS));
|
|
const bool serialize_init =
|
|
uhd::cast::from_str<bool>(device_args.get("serialize_init", "0"))
|
|
|| max_init_threads == 1;
|
|
const bool skip_init = uhd::cast::from_str<bool>(device_args.get("skip_init", "0"));
|
|
if (serialize_init) {
|
|
UHD_LOG_INFO("MPMD",
|
|
"Initializing " << num_mboards << " device(s) serially with args: "
|
|
<< device_args.to_string());
|
|
} else {
|
|
UHD_LOG_WARNING("MPMD",
|
|
"Parallel initialization has not been validated for all USRPs. Use at your "
|
|
"own risk.");
|
|
UHD_LOG_INFO("MPMD",
|
|
"Initializing " << num_mboards
|
|
<< " device(s) in parallel (max threads: " << max_init_threads
|
|
<< ") at a time, with args : " << device_args.to_string());
|
|
}
|
|
|
|
// Helper to run a per-mboard task across all motherboards in parallel,
|
|
// using at most max_init_threads threads at a time so we don't spawn
|
|
// an unbounded number of threads for large device counts.
|
|
// All tasks in a batch are awaited before any exception is re-thrown, so
|
|
// no async task can outlive this scope and access a destroyed or
|
|
// partially-constructed object.
|
|
auto run_parallel = [num_mboards, max_init_threads](auto&& task) {
|
|
size_t mb_i = 0;
|
|
while (mb_i < num_mboards) {
|
|
const size_t batch_size = std::min(num_mboards - mb_i, max_init_threads);
|
|
std::vector<std::future<void>> futures;
|
|
futures.reserve(batch_size);
|
|
for (size_t i = 0; i < batch_size; ++i) {
|
|
const size_t index = mb_i + i;
|
|
futures.push_back(
|
|
std::async(std::launch::async, [&task, index]() { task(index); }));
|
|
}
|
|
// Wait for every task in this batch to finish before propagating
|
|
// any exception, so no task is left running while we unwind.
|
|
std::exception_ptr first_exception;
|
|
for (auto& future : futures) {
|
|
try {
|
|
future.get();
|
|
} catch (...) {
|
|
if (!first_exception) {
|
|
first_exception = std::current_exception();
|
|
} else {
|
|
// We can only re-throw one exception, so log any
|
|
// additional ones so they're not silently swallowed.
|
|
try {
|
|
std::rethrow_exception(std::current_exception());
|
|
} catch (const std::exception& ex) {
|
|
UHD_LOG_ERROR("MPMD",
|
|
"Additional exception during parallel "
|
|
"initialization: "
|
|
<< ex.what());
|
|
} catch (...) {
|
|
UHD_LOG_ERROR("MPMD",
|
|
"Additional unknown exception during parallel "
|
|
"initialization.");
|
|
}
|
|
}
|
|
}
|
|
}
|
|
if (first_exception) {
|
|
std::rethrow_exception(first_exception);
|
|
}
|
|
mb_i += batch_size;
|
|
}
|
|
};
|
|
|
|
// 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.
|
|
if (serialize_init) {
|
|
for (size_t mb_i = 0; mb_i < num_mboards; ++mb_i) {
|
|
UHD_LOG_DEBUG("MPMD", "Claiming mboard " << mb_i);
|
|
_mb[mb_i] = claim_and_make(mb_args[mb_i], mb_i);
|
|
}
|
|
} else {
|
|
run_parallel([this, &mb_args](const size_t mb_i) {
|
|
UHD_LOG_DEBUG("MPMD", "Claiming mboard " << mb_i);
|
|
_mb[mb_i] = claim_and_make(mb_args[mb_i], mb_i);
|
|
});
|
|
}
|
|
|
|
if (not skip_init) {
|
|
// Run the actual device initialization. This can run in parallel.
|
|
// Note: setup_mb() only performs the (thread-safe) heavy init; the
|
|
// motherboard controllers are registered serially below, because
|
|
// register_mb_controller() mutates a shared registry that is not
|
|
// thread-safe.
|
|
if (serialize_init) {
|
|
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);
|
|
}
|
|
} else {
|
|
run_parallel([this](const size_t mb_i) { setup_mb(_mb[mb_i].get(), mb_i); });
|
|
}
|
|
// Register the motherboard controllers serially: this mutates a shared
|
|
// registry that is not safe for concurrent access.
|
|
for (size_t mb_i = 0; mb_i < num_mboards; ++mb_i) {
|
|
UHD_ASSERT_THROW(_mb[mb_i]->mb_ctrl);
|
|
register_mb_controller(mb_i, _mb[mb_i]->mb_ctrl);
|
|
}
|
|
} else {
|
|
UHD_LOG_DEBUG("MPMD", "Claimed device(s), 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(), mb_i);
|
|
}
|
|
|
|
if (not skip_init) {
|
|
// FIXME this section only makes sense for when the time source is external.
|
|
// So, check for that, or something similar.
|
|
// This section of code assumes that the prop tree is set and we have access
|
|
// to the timekeepers. So don't move it anywhere else.
|
|
if (device_args.has_key("sync_time")) {
|
|
reset_time_synchronized(_tree);
|
|
}
|
|
} else {
|
|
UHD_LOG_INFO("MPMD", "Claimed device without full initialization.");
|
|
}
|
|
}
|
|
|
|
mpmd_impl::~mpmd_impl()
|
|
{
|
|
_deinit();
|
|
}
|
|
|
|
/*****************************************************************************
|
|
* Protected methods
|
|
****************************************************************************/
|
|
void mpmd_impl::_deinit()
|
|
{
|
|
_tree.reset();
|
|
_mb.clear();
|
|
}
|
|
|
|
/*****************************************************************************
|
|
* Private methods
|
|
****************************************************************************/
|
|
mpmd_mboard_impl::uptr mpmd_impl::claim_and_make(
|
|
const uhd::device_addr_t& device_args, const size_t mb_index)
|
|
{
|
|
const std::string rpc_addr = device_args.get(MGMT_ADDR_KEY);
|
|
UHD_LOGGER_DEBUG("MPMD") << "Device args: `" << device_args.to_string()
|
|
<< "'. RPC address: " << rpc_addr;
|
|
|
|
if (rpc_addr.empty()) {
|
|
UHD_LOG_ERROR("MPMD",
|
|
"Could not determine RPC address from device args: "
|
|
<< device_args.to_string());
|
|
throw uhd::runtime_error("Could not determine device RPC address.");
|
|
}
|
|
return mpmd_mboard_impl::make(device_args, rpc_addr, mb_index);
|
|
}
|
|
|
|
void mpmd_impl::setup_mb(mpmd_mboard_impl* mb, const size_t mb_index)
|
|
{
|
|
auto compat_num_uint32 = mb->rpc->get_mpm_compat_number();
|
|
std::vector<size_t> compat_num(compat_num_uint32.begin(), compat_num_uint32.end());
|
|
assert_compat_number_throw("MPM",
|
|
MPM_COMPAT_NUM,
|
|
compat_num,
|
|
"Please update the version of MPM on your USRP device.");
|
|
|
|
UHD_LOG_DEBUG("MPMD", "Initializing mboard " << mb_index);
|
|
mb->init();
|
|
UHD_ASSERT_THROW(mb->mb_ctrl);
|
|
}
|
|
|
|
/*****************************************************************************
|
|
* Factory & Registry
|
|
****************************************************************************/
|
|
static device::sptr mpmd_make(const device_addr_t& device_args)
|
|
{
|
|
return device::sptr(std::make_shared<mpmd_impl>(device_args));
|
|
}
|
|
|
|
UHD_STATIC_BLOCK(register_mpmd_device)
|
|
{
|
|
device::register_device(&mpmd_find, &mpmd_make, device::USRP);
|
|
}
|
|
// vim: sw=4 expandtab:
|