// // Copyright 2019 Ettus Research, a National Instruments Brand // // SPDX-License-Identifier: GPL-3.0-or-later // #include #include #include #include #include #include #include using namespace uhd::rfnoc; dirtifier_t node_t::ALWAYS_DIRTY{}; node_t::node_t() { register_property(&ALWAYS_DIRTY); } std::string node_t::get_unique_id() const { // TODO return something better return str(boost::format("%08X") % this); } std::vector node_t::get_property_ids() const { std::lock_guard _l(_prop_mutex); if (_props.count(res_source_info::USER) == 0) { return {}; } auto& user_props = _props.at(res_source_info::USER); // TODO use a range here, we're not savages std::vector return_value(user_props.size()); for (size_t i = 0; i < user_props.size(); ++i) { return_value[i] = user_props[i]->get_id(); } return return_value; } void node_t::set_properties(const uhd::device_addr_t& props, const size_t instance) { for (const auto& key : props.keys()) { property_base_t* prop_ref = _find_property({res_source_info::USER, instance}, key); if (!prop_ref) { RFNOC_LOG_WARNING("set_properties() cannot set property `" << key << "': No such property."); continue; } auto prop_access = _request_property_access(prop_ref, property_base_t::RW); prop_ref->set_from_str(props.get(key)); } // Now trigger a property resolution. If other properties depend on modified // properties, they will be updated. resolve_all(); } void node_t::set_command_time(uhd::time_spec_t time, const size_t instance) { if (_cmd_timespecs.size() <= instance) { _cmd_timespecs.resize(instance + 1, uhd::time_spec_t(0.0)); } _cmd_timespecs[instance] = time; } uhd::time_spec_t node_t::get_command_time(const size_t instance) const { if (instance >= _cmd_timespecs.size()) { return uhd::time_spec_t::ASAP; } return _cmd_timespecs.at(instance); } void node_t::clear_command_time(const size_t instance) { set_command_time(uhd::time_spec_t(0.0), instance); } /*** Protected methods *******************************************************/ void node_t::register_property(property_base_t* prop, resolve_callback_t&& clean_callback) { std::lock_guard _l(_prop_mutex); const auto src_type = prop->get_src_info().type; // If the map is empty for this source type, create an empty vector if (_props.count(src_type) == 0) { _props[src_type] = {}; } auto prop_already_registered = [prop](const property_base_t* existing_prop) { return (prop == existing_prop) || (prop->get_src_info() == existing_prop->get_src_info() && prop->get_id() == existing_prop->get_id()); }; if (!filter_props(prop_already_registered).empty()) { throw uhd::runtime_error(std::string("Attempting to double-register property: ") + prop->get_id() + "[" + prop->get_src_info().to_string() + "]"); } _props[src_type].push_back(prop); if (clean_callback) { _clean_cb_registry[prop] = std::move(clean_callback); } prop_accessor_t{}.set_access(prop, property_base_t::RW); } void node_t::add_property_resolver( prop_ptrs_t&& inputs, prop_ptrs_t&& outputs, resolver_fn_t&& resolver_fn) { std::lock_guard _l(_prop_mutex); // Sanity check: All inputs and outputs must be registered properties auto prop_is_registered = [this](property_base_t* prop) -> bool { return bool(this->_find_property(prop->get_src_info(), prop->get_id())); }; for (const auto& prop : inputs) { if (!prop_is_registered(prop)) { throw uhd::runtime_error( std::string("Cannot add property resolver, input property ") + prop->get_id() + " is not registered!"); } } for (const auto& prop : outputs) { if (!prop_is_registered(prop)) { throw uhd::runtime_error( std::string("Cannot add property resolver, output property ") + prop->get_id() + " is not registered!"); } } // All good, we can store it _prop_resolvers.push_back(std::make_tuple(std::forward(inputs), std::forward(outputs), std::forward(resolver_fn))); } void node_t::set_prop_forwarding_policy( forwarding_policy_t policy, const std::string& prop_id) { _prop_fwd_policies[prop_id] = policy; } void node_t::register_action_handler(const std::string& id, action_handler_t&& handler) { if (_action_handlers.count(id)) { _action_handlers.erase(id); } _action_handlers.emplace(id, std::move(handler)); } void node_t::set_action_forwarding_policy( node_t::forwarding_policy_t policy, const std::string& action_key) { _action_fwd_policies[action_key] = policy; } void node_t::post_action( const res_source_info& edge_info, action_info::sptr action) { _post_action_cb(edge_info, action); } bool node_t::check_topology(const std::vector& connected_inputs, const std::vector& connected_outputs) { for (size_t port : connected_inputs) { if (port >= get_num_input_ports()) { return false; } } for (size_t port : connected_outputs) { if (port >= get_num_output_ports()) { return false; } } return true; } /*** Private methods *********************************************************/ property_base_t* node_t::_find_property( res_source_info src_info, const std::string& id) const { for (const auto& type_prop_pair : _props) { if (type_prop_pair.first != src_info.type) { continue; } for (const auto& prop : type_prop_pair.second) { if (prop->get_id() == id && prop->get_src_info() == src_info) { return prop; } } } return nullptr; } uhd::utils::scope_exit::uptr node_t::_request_property_access( property_base_t* prop, property_base_t::access_t access) const { return prop_accessor_t{}.get_scoped_prop_access(*prop, access); } property_base_t* node_t::inject_edge_property( property_base_t* blueprint, res_source_info new_src_info) { // Check if a property already exists which matches the new property // requirements. If so, we can return early: auto new_prop = _find_property(new_src_info, blueprint->get_id()); if (new_prop) { return new_prop; } // We need to create a new property and stash it away: new_prop = [&]() -> property_base_t* { auto prop = blueprint->clone(new_src_info); auto ptr = prop.get(); _dynamic_props.emplace(std::move(prop)); return ptr; }(); register_property(new_prop); // Collect some info on how to do the forwarding: const auto fwd_policy = [&](const std::string& id) { if (_prop_fwd_policies.count(id)) { return _prop_fwd_policies.at(id); } return _prop_fwd_policies.at(""); }(new_prop->get_id()); const size_t port_idx = new_prop->get_src_info().instance; const auto port_type = new_prop->get_src_info().type; UHD_ASSERT_THROW(port_type == res_source_info::INPUT_EDGE || port_type == res_source_info::OUTPUT_EDGE); // Now comes the hard part: Figure out which other properties need to be // created, and which resolvers need to be instantiated if (fwd_policy == forwarding_policy_t::ONE_TO_ONE) { // Figure out if there's an opposite port const auto opposite_port_type = res_source_info::invert_edge(port_type); if (_has_port({opposite_port_type, port_idx})) { // Make sure that the other side's property exists: // This is a safe recursion, because we've already created and // registered this property. auto opposite_prop = inject_edge_property(new_prop, {opposite_port_type, port_idx}); // Now add a resolver that will always forward the value from this // property to the other one. add_property_resolver( {new_prop}, {opposite_prop}, [new_prop, opposite_prop]() { prop_accessor_t{}.forward(new_prop, opposite_prop); }); } } if (fwd_policy == forwarding_policy_t::ONE_TO_FAN) { const auto opposite_port_type = res_source_info::invert_edge(port_type); const size_t num_ports = opposite_port_type == res_source_info::INPUT_EDGE ? get_num_input_ports() : get_num_output_ports(); for (size_t i = 0; i < num_ports; i++) { auto opposite_prop = inject_edge_property(new_prop, {opposite_port_type, i}); // Now add a resolver that will always forward the value from this // property to the other one. add_property_resolver( {new_prop}, {opposite_prop}, [new_prop, opposite_prop]() { prop_accessor_t{}.forward(new_prop, opposite_prop); }); } } if (fwd_policy == forwarding_policy_t::ONE_TO_ALL || fwd_policy == forwarding_policy_t::ONE_TO_ALL_IN) { // Loop through all other ports, make sure those properties exist for (size_t other_port_idx = 0; other_port_idx < get_num_input_ports(); other_port_idx++) { if (port_type == res_source_info::INPUT_EDGE && other_port_idx == port_idx) { continue; } inject_edge_property(new_prop, {res_source_info::INPUT_EDGE, other_port_idx}); } // Now add a dynamic resolver that will update all input properties. // In order to keep this code simple, we bypass the write list and // get access via the prop_accessor. add_property_resolver({new_prop}, {/* empty */}, [this, new_prop, port_idx]() { for (size_t other_port_idx = 0; other_port_idx < get_num_input_ports(); other_port_idx++) { if (other_port_idx == port_idx) { continue; } auto prop = _find_property({res_source_info::INPUT_EDGE, other_port_idx}, new_prop->get_id()); if (prop) { prop_accessor_t{}.forward(new_prop, prop); } } }); } if (fwd_policy == forwarding_policy_t::ONE_TO_ALL || fwd_policy == forwarding_policy_t::ONE_TO_ALL_OUT) { // Loop through all other ports, make sure those properties exist for (size_t other_port_idx = 0; other_port_idx < get_num_output_ports(); other_port_idx++) { if (port_type == res_source_info::OUTPUT_EDGE && other_port_idx == port_idx) { continue; } inject_edge_property(new_prop, {res_source_info::OUTPUT_EDGE, other_port_idx}); } // Now add a dynamic resolver that will update all input properties. // In order to keep this code simple, we bypass the write list and // get access via the prop_accessor. add_property_resolver({new_prop}, {/* empty */}, [this, new_prop, port_idx]() { for (size_t other_port_idx = 0; other_port_idx < get_num_input_ports(); other_port_idx++) { if (other_port_idx == port_idx) { continue; } auto prop = _find_property({res_source_info::OUTPUT_EDGE, other_port_idx}, new_prop->get_id()); if (prop) { prop_accessor_t{}.forward(new_prop, prop); } } }); } return new_prop; } void node_t::init_props() { std::lock_guard _l(_prop_mutex); prop_accessor_t prop_accessor{}; for (auto& resolver_tuple : _prop_resolvers) { // 1) Set all outputs to RWLOCKED auto& outputs = std::get<1>(resolver_tuple); for (auto& output : outputs) { prop_accessor.set_access(output, property_base_t::RWLOCKED); } // 2) Run the resolver try { std::get<2>(resolver_tuple)(); } catch (const uhd::resolve_error& ex) { UHD_LOGGER_WARNING(get_unique_id()) << "Failed to initialize node. Most likely cause: Inconsistent default " "values. Resolver threw this error: " << ex.what(); //throw uhd::runtime_error(std::string("Failed to initialize node ") + get_unique_id()); } // 3) Set outputs back to RO for (auto& output : outputs) { prop_accessor.set_access(output, property_base_t::RO); } } // 4) Mark properties as clean and read-only clean_props(); } void node_t::resolve_props() { prop_accessor_t prop_accessor{}; const prop_ptrs_t initial_dirty_props = filter_props([](property_base_t* prop) { return prop->is_dirty(); }); std::list all_dirty_props( initial_dirty_props.cbegin(), initial_dirty_props.cend()); prop_ptrs_t processed_props{}; prop_ptrs_t written_props{}; RFNOC_LOG_TRACE("Locally resolving " << all_dirty_props.size() << " dirty properties plus dependencies."); // Loop through all dirty properties. The list can be amended during the // loop execution. for (auto it = all_dirty_props.begin(); it != all_dirty_props.end(); ++it) { auto current_input_prop = *it; if (processed_props.count(current_input_prop)) { continue; } // Find all resolvers that take this dirty property as an input: for (auto& resolver_tuple : _prop_resolvers) { auto& inputs = std::get<0>(resolver_tuple); auto& outputs = std::get<1>(resolver_tuple); if (!inputs.count(current_input_prop)) { continue; } // Enable outputs std::vector access_holder; access_holder.reserve(outputs.size()); for (auto& output : outputs) { access_holder.emplace_back(prop_accessor.get_scoped_prop_access(*output, written_props.count(output) ? property_base_t::access_t::RWLOCKED : property_base_t::access_t::RW)); } // Run resolver std::get<2>(resolver_tuple)(); // Take note of outputs written_props.insert(outputs.cbegin(), outputs.cend()); // Add all outputs that are dirty to the list, unless they have // already been processed for (auto& output_prop : outputs) { if (output_prop->is_dirty() && processed_props.count(output_prop) == 0) { all_dirty_props.push_back(output_prop); } } // RW or RWLOCKED gets released here as access_holder goes out of scope. } processed_props.insert(current_input_prop); } } void node_t::resolve_all() { _resolve_all_cb(); } void node_t::clean_props() { prop_accessor_t prop_accessor{}; for (const auto& type_prop_pair : _props) { for (const auto& prop : type_prop_pair.second) { if (prop->is_dirty() && _clean_cb_registry.count(prop)) { _clean_cb_registry.at(prop)(); } prop_accessor.mark_clean(*prop); prop_accessor.set_access(prop, property_base_t::RO); } } } void node_t::forward_edge_property( property_base_t* incoming_prop, const size_t incoming_port) { UHD_ASSERT_THROW( incoming_prop->get_src_info().type == res_source_info::INPUT_EDGE || incoming_prop->get_src_info().type == res_source_info::OUTPUT_EDGE); UHD_LOG_TRACE(get_unique_id(), "Incoming edge property: `" << incoming_prop->get_id() << "`, source info: " << incoming_prop->get_src_info().to_string()); // Don't forward properties that are not yet valid if (!incoming_prop->is_valid()) { UHD_LOG_TRACE(get_unique_id(), "Skipped empty edge property: `" << incoming_prop->get_id() << "`, source info: " << incoming_prop->get_src_info().to_string()); return; } // The source type of my local prop (it's the opposite of the source type // of incoming_prop) const auto prop_src_type = res_source_info::invert_edge(incoming_prop->get_src_info().type); // Set of local properties that match incoming_prop. It can be an empty set, // or, if the node is misconfigured, a set with more than one entry. Or, if // all is as expected, it's a set with a single entry. auto local_prop_set = filter_props([prop_src_type, incoming_prop, incoming_port]( property_base_t* prop) -> bool { return prop->get_src_info().type == prop_src_type && prop->get_src_info().instance == incoming_port && prop->get_id() == incoming_prop->get_id(); }); // If there is no such property, we're forwarding a new property if (local_prop_set.empty()) { UHD_LOG_TRACE(get_unique_id(), "Received unknown incoming edge prop: " << incoming_prop->get_id()); local_prop_set.emplace( inject_edge_property(incoming_prop, {prop_src_type, incoming_port})); } // There must be either zero results, or one UHD_ASSERT_THROW(local_prop_set.size() == 1); auto local_prop = *local_prop_set.begin(); prop_accessor_t prop_accessor{}; prop_accessor.forward(incoming_prop, local_prop); } void node_t::receive_action(const res_source_info& src_info, action_info::sptr action) { std::lock_guard l(_action_mutex); // See if the user defined an action handler for us: if (_action_handlers.count(action->key)) { _action_handlers.at(action->key)(src_info, action); return; } // We won't forward actions if they were for us if (src_info.type == res_source_info::USER) { RFNOC_LOG_TRACE("Dropping USER action " << action->key << "#" << action->id); return; } // Otherwise, we need to figure out the correct default action handling: const auto fwd_policy = [&](const std::string& id) { if (_action_fwd_policies.count(id)) { return _action_fwd_policies.at(id); } return _action_fwd_policies.at(""); }(action->key); // Now implement custom forwarding for all forwarding policies: if (fwd_policy == forwarding_policy_t::DROP) { UHD_LOG_TRACE( get_unique_id(), "Dropping action " << action->key); } if (fwd_policy == forwarding_policy_t::ONE_TO_ONE) { UHD_LOG_TRACE( get_unique_id(), "Forwarding action " << action->key << " to opposite port"); const res_source_info dst_info{ res_source_info::invert_edge(src_info.type), src_info.instance}; if (_has_port(dst_info)) { post_action(dst_info, action); } } if (fwd_policy == forwarding_policy_t::ONE_TO_FAN) { UHD_LOG_TRACE(get_unique_id(), "Forwarding action " << action->key << " to all opposite ports"); const auto new_edge_type = res_source_info::invert_edge(src_info.type); const size_t num_ports = new_edge_type == res_source_info::INPUT_EDGE ? get_num_input_ports() : get_num_output_ports(); for (size_t i = 0; i < num_ports; i++) { post_action({new_edge_type, i}, action); } } if (fwd_policy == forwarding_policy_t::ONE_TO_ALL || fwd_policy == forwarding_policy_t::ONE_TO_ALL_IN) { UHD_LOG_TRACE(get_unique_id(), "Forwarding action " << action->key << " to all input ports"); for (size_t i = 0; i < get_num_input_ports(); i++) { if (src_info.type == res_source_info::INPUT_EDGE && i == src_info.instance) { continue; } post_action({res_source_info::INPUT_EDGE, i}, action); } } if (fwd_policy == forwarding_policy_t::ONE_TO_ALL || fwd_policy == forwarding_policy_t::ONE_TO_ALL_OUT) { UHD_LOG_TRACE(get_unique_id(), "Forwarding action " << action->key << " to all output ports"); for (size_t i = 0; i < get_num_output_ports(); i++) { if (src_info.type == res_source_info::OUTPUT_EDGE && i == src_info.instance) { continue; } post_action({res_source_info::OUTPUT_EDGE, i}, action); } } } void node_t::shutdown() { RFNOC_LOG_DEBUG("shutdown() not implemented."); } bool node_t::_has_port(const res_source_info& port_info) const { return (port_info.type == res_source_info::INPUT_EDGE && port_info.instance <= get_num_input_ports()) || (port_info.type == res_source_info::OUTPUT_EDGE && port_info.instance <= get_num_output_ports()); }