diff options
| -rw-r--r-- | host/include/uhd/rfnoc/replay_block_control.hpp | 43 | ||||
| -rw-r--r-- | host/lib/rfnoc/replay_block_control.cpp | 72 | ||||
| -rw-r--r-- | host/tests/rfnoc_block_tests/replay_block_test.cpp | 17 |
3 files changed, 130 insertions, 2 deletions
diff --git a/host/include/uhd/rfnoc/replay_block_control.hpp b/host/include/uhd/rfnoc/replay_block_control.hpp index a921d1bca..615f6e294 100644 --- a/host/include/uhd/rfnoc/replay_block_control.hpp +++ b/host/include/uhd/rfnoc/replay_block_control.hpp @@ -95,6 +95,17 @@ namespace uhd { namespace rfnoc { * * To stop a continuous playback, either call stop(), or issue a stream command * with uhd::stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS. + * + * \section rfnoc_block_replay_actions Action Handling + * + * If this block receives TX or RX actions (uhd::rfnoc::tx_event_action_info o + * uhd::rfnoc::rx_event_action_info), it will store them in a circular buffer. + * The API calls get_record_async_metadata() and get_play_async_metadata() can + * be used to read them back out asynchronously. To avoid the block controller + * continously expanding in memory, the total number of messages that will be + * stored is limited. If this block receives more event info objects than it can + * store before get_record_async_metadata() or get_play_async_metadata() is + * called, the oldest message will be dropped. */ // clang-format on class UHD_API replay_block_control : public noc_block_base @@ -273,6 +284,22 @@ public: */ virtual size_t get_record_item_size(const size_t port = 0) const = 0; + /*! Return RX- (input-/record-) related metadata. + * + * The typical use case for this is when connecting Radio -> Replay for + * recording, the radio may produce information like 'overrun occurred'. + * When receiving to a host using a uhd::rx_streamer, this information is + * returned as part of the uhd::rx_streamer::recv() call, but when the data + * is streamed into the replay block, these metadata are stored inside the + * replay block until queried by this method. + * + * \param metadata A metadata object to store the information in. + * \param timeout A timeout (in seconds) to wait before returning. + * \returns true if a message was available, and was popped into \p metadata. + */ + virtual bool get_record_async_metadata( + uhd::rx_metadata_t& metadata, const double timeout = 0.1) = 0; + /************************************************************************** * Playback State API calls *************************************************************************/ @@ -320,6 +347,22 @@ public: */ virtual size_t get_play_item_size(const size_t port = 0) const = 0; + /*! Return TX- (output-/playback-) related metadata. + * + * The typical use case for this is when connecting Replay -> Radio for + * playback, the radio may produce information like 'underrun occurred'. + * When transmitting from a host using a uhd::tx_streamer, this information + * is returned as part of the uhd::tx_streamer::recv_async_msg() call, but + * when the data is streamed into the replay block, these metadata are + * stored inside the replay block until queried by this method. + * + * \param metadata A metadata object to store the information in. + * \param timeout A timeout (in seconds) to wait before returning. + * \returns true if a message was available, and was popped into \p metadata. + */ + virtual bool get_play_async_metadata( + uhd::async_metadata_t& metadata, const double timeout = 0.1) = 0; + /************************************************************************** * Advanced Record Control API calls *************************************************************************/ diff --git a/host/lib/rfnoc/replay_block_control.cpp b/host/lib/rfnoc/replay_block_control.cpp index b28476ea8..2ee31de93 100644 --- a/host/lib/rfnoc/replay_block_control.cpp +++ b/host/lib/rfnoc/replay_block_control.cpp @@ -11,6 +11,7 @@ #include <uhd/rfnoc/property.hpp> #include <uhd/rfnoc/registry.hpp> #include <uhd/rfnoc/replay_block_control.hpp> +#include <uhd/transport/bounded_buffer.hpp> #include <uhd/types/stream_cmd.hpp> #include <uhd/utils/math.hpp> #include <uhdlib/utils/compat_check.hpp> @@ -67,6 +68,9 @@ const char* const PROP_KEY_PLAY_OFFSET = "play_offset"; const char* const PROP_KEY_PLAY_SIZE = "play_size"; const char* const PROP_KEY_PKT_SIZE = "packet_size"; +// Depth of the async message queues +constexpr size_t ASYNC_MSG_QUEUE_SIZE = 128; + class replay_block_control_impl : public replay_block_control { public: @@ -124,6 +128,26 @@ public: } issue_stream_cmd(stream_cmd_action->stream_cmd, port); }); + register_action_handler(ACTION_KEY_RX_EVENT, + [this](const res_source_info& src, action_info::sptr action) { + rx_event_action_info::sptr rx_event_action = + std::dynamic_pointer_cast<rx_event_action_info>(action); + if (!rx_event_action) { + RFNOC_LOG_WARNING("Received invalid RX event action!"); + return; + } + _handle_rx_event_action(src, rx_event_action); + }); + register_action_handler(ACTION_KEY_TX_EVENT, + [this](const res_source_info& src, action_info::sptr action) { + tx_event_action_info::sptr tx_event_action = + std::dynamic_pointer_cast<tx_event_action_info>(action); + if (!tx_event_action) { + RFNOC_LOG_WARNING("Received invalid TX event action!"); + return; + } + _handle_tx_event_action(src, tx_event_action); + }); // Initialize record properties _record_type.reserve(_num_input_ports); @@ -242,6 +266,12 @@ public: return uhd::convert::get_bytes_per_item(get_record_type(port)); } + bool get_record_async_metadata( + uhd::rx_metadata_t& metadata, const double timeout = 0.0) + { + return _record_msg_queue.pop_with_timed_wait(metadata, timeout); + } + /************************************************************************** * Playback State API **************************************************************************/ @@ -276,6 +306,11 @@ public: return uhd::convert::get_bytes_per_item(get_play_type(port)); } + bool get_play_async_metadata(uhd::async_metadata_t& metadata, const double timeout) + { + return _playback_msg_queue.pop_with_timed_wait(metadata, timeout); + } + /************************************************************************** * Advanced Record Control API calls *************************************************************************/ @@ -583,6 +618,37 @@ private: } } + void _handle_rx_event_action( + const res_source_info& src, rx_event_action_info::sptr rx_event_action) + { + UHD_ASSERT_THROW(src.type == res_source_info::INPUT_EDGE); + uhd::rx_metadata_t rx_md{}; + rx_md.error_code = rx_event_action->error_code; + RFNOC_LOG_DEBUG("Received RX error code on channel " + << src.instance << ", error code " << rx_md.strerror()); + _record_msg_queue.push_with_pop_on_full(rx_md); + } + + void _handle_tx_event_action( + const res_source_info& src, tx_event_action_info::sptr tx_event_action) + { + UHD_ASSERT_THROW(src.type == res_source_info::OUTPUT_EDGE); + + uhd::async_metadata_t md; + md.event_code = tx_event_action->event_code; + md.channel = src.instance; + md.has_time_spec = tx_event_action->has_tsf; + + if (md.has_time_spec) { + md.time_spec = + uhd::time_spec_t::from_ticks(tx_event_action->tsf, get_tick_rate()); + } + RFNOC_LOG_DEBUG("Received TX error code on channel " + << src.instance << ", error code " + << static_cast<int>(md.event_code)); + _playback_msg_queue.push_with_pop_on_full(md); + } + /************************************************************************** * Attributes *************************************************************************/ @@ -605,6 +671,12 @@ private: std::vector<property_t<uint32_t>> _packet_size; std::vector<property_t<size_t>> _atomic_item_size_in; std::vector<property_t<size_t>> _atomic_item_size_out; + + // Message queues for async data + uhd::transport::bounded_buffer<uhd::async_metadata_t> _playback_msg_queue{ + ASYNC_MSG_QUEUE_SIZE}; + uhd::transport::bounded_buffer<uhd::rx_metadata_t> _record_msg_queue{ + ASYNC_MSG_QUEUE_SIZE}; }; UHD_RFNOC_BLOCK_REGISTER_DIRECT( diff --git a/host/tests/rfnoc_block_tests/replay_block_test.cpp b/host/tests/rfnoc_block_tests/replay_block_test.cpp index 11c621b5a..94f1ce882 100644 --- a/host/tests/rfnoc_block_tests/replay_block_test.cpp +++ b/host/tests/rfnoc_block_tests/replay_block_test.cpp @@ -749,8 +749,6 @@ BOOST_FIXTURE_TEST_CASE(replay_test_graph, replay_block_fixture) node_accessor.init_props(&mock_ddc_block); mock_sink_term.set_edge_property<std::string>( "type", "sc16", {res_source_info::INPUT_EDGE, 0}); - mock_sink_term.set_edge_property<std::string>( - "type", "sc16", {res_source_info::INPUT_EDGE, 1}); UHD_LOG_INFO("TEST", "Creating graph..."); graph.connect(&mock_radio_block, &mock_ddc_block, edge_port_info); @@ -758,7 +756,22 @@ BOOST_FIXTURE_TEST_CASE(replay_test_graph, replay_block_fixture) graph.connect(test_replay.get(), &mock_sink_term, edge_port_info); UHD_LOG_INFO("TEST", "Committing graph..."); graph.commit(); + mock_sink_term.set_edge_property<double>( + "tick_rate", 1.0, {res_source_info::INPUT_EDGE, 0}); UHD_LOG_INFO("TEST", "Commit complete."); + + mock_radio_block.generate_overrun(0); + uhd::rx_metadata_t rx_md; + BOOST_REQUIRE(test_replay->get_record_async_metadata(rx_md, 1.0)); + BOOST_CHECK(rx_md.error_code == uhd::rx_metadata_t::ERROR_CODE_OVERFLOW); + + mock_sink_term.post_action(res_source_info{res_source_info::INPUT_EDGE, 0}, + tx_event_action_info::make(uhd::async_metadata_t::EVENT_CODE_UNDERFLOW, 1234ul)); + uhd::async_metadata_t tx_md; + BOOST_REQUIRE(test_replay->get_play_async_metadata(tx_md, 1.0)); + BOOST_CHECK(tx_md.event_code == uhd::async_metadata_t::EVENT_CODE_UNDERFLOW); + BOOST_CHECK(tx_md.has_time_spec); + BOOST_CHECK(tx_md.time_spec == 1234.0); } /* |