rfnoc: replay: Add ability to capture and read async info

- Add action handlers to the replay block to store TX and RX events.
- Adds two new APIs: get_{record,play}_async_metadata() to read back
  async info.
- Add unit tests.

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);
 }
 
 /*