rfnoc: Implement flushing on overrun

This modifies the overrun handling such that the RX streamer does not
restart the radios until the packets that were buffered prior to the
overrun are read by the user.

When an RX streamer receives an overrun, it will run the following
algorithm:
1. Stop all upstream producers.
2. Set an internal flag in the streamer that indicates that the
   producers have stopped due to an overrun.
3. Continue servicing calls to recv until it runs out of packets in the
   host buffer (packets that can be read from the transport using a 0
   timeout).
4. Once the packets are exhausted, return an overrun error from recv.

The radio, if it was in continuous streaming mode before the overrun,
includes a flag in its initial action whether or not to restart
streaming.

5. If the radio requested a restart, the streamer submits a restart
   request action upstream. This action will be received by the radio.
   The radio will then check the current time, and send a stream command
   action back downstream.
6. The RX streamer receives the stream command action, and uses it to
   send another stream command to all upstream producers. This way, all
   upstream producers receive a start command for the same time.

2 files changed, 28 insertions, 43 deletions

diff --git a/host/lib/rfnoc/radio_control_impl.cpp b/host/lib/rfnoc/radio_control_impl.cpp
index 9d7257108..e400033b3 100644
--- a/host/lib/rfnoc/radio_control_impl.cpp
+++ b/host/lib/rfnoc/radio_control_impl.cpp
@@ -6,6 +6,7 @@
 
 #include <uhd/exception.hpp>
 #include <uhd/utils/log.hpp>
+#include <uhd/rfnoc/mb_controller.hpp>
 #include <uhdlib/rfnoc/radio_control_impl.hpp>
 #include <uhdlib/utils/compat_check.hpp>
 #include <map>
@@ -63,6 +64,8 @@ const uint32_t radio_control_impl::regmap::RX_CMD_TIMED_POS;
 const uhd::fs_path radio_control_impl::DB_PATH("dboard");
 const uhd::fs_path radio_control_impl::FE_PATH("frontends");
 
+static constexpr double OVERRUN_RESTART_DELAY = 0.05;
+
 /****************************************************************************
  * Structors
  ***************************************************************************/
@@ -74,7 +77,6 @@ radio_control_impl::radio_control_impl(make_args_ptr make_args)
     , _spc(_radio_width & 0xFFFF)
     , _last_stream_cmd(
           get_num_output_ports(), uhd::stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS)
-    , _restart_cont(get_num_output_ports(), false)
 {
     uhd::assert_fpga_compat(MAJOR_COMPAT,
         MINOR_COMPAT,
@@ -110,16 +112,6 @@ radio_control_impl::radio_control_impl(make_args_ptr make_args)
                 return;
             }
             issue_stream_cmd(stream_cmd_action->stream_cmd, port);
-            if (stream_cmd_action->stream_cmd.stream_mode
-                    == uhd::stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS
-                && _restart_cont.at(port)) {
-                RFNOC_LOG_TRACE("Received stop command after reporting overrun, will now "
-                                "request restart.");
-                _restart_cont[port] = false;
-                auto restart_request_action =
-                    action_info::make(ACTION_KEY_RX_RESTART_REQ);
-                post_action({res_source_info::OUTPUT_EDGE, port}, restart_request_action);
-            }
         });
     register_action_handler(ACTION_KEY_RX_RESTART_REQ,
         [this](const res_source_info& src, action_info::sptr /*action*/) {
@@ -131,9 +123,10 @@ radio_control_impl::radio_control_impl(make_args_ptr make_args)
             }
             auto stream_cmd_action = stream_cmd_action_info::make(
                 uhd::stream_cmd_t::STREAM_MODE_START_CONTINUOUS);
-            // FIXME
-            // stream_cmd_action->stream_cmd.stream_now = false;
-            // stream_cmd_action->stream_cmd.time_spec = get_time_now() + DELTA;
+            stream_cmd_action->stream_cmd.stream_now = false;
+            stream_cmd_action->stream_cmd.time_spec =
+                get_mb_controller()->get_timekeeper(0)->get_time_now() +
+                uhd::time_spec_t(OVERRUN_RESTART_DELAY);
             const size_t port = src.instance;
             if (port > get_num_output_ports()) {
                 RFNOC_LOG_WARNING("Received stream command to invalid output port!");
@@ -962,7 +955,6 @@ void radio_control_impl::async_message_handler(
                     rx_event_action->error_code = uhd::rx_metadata_t::ERROR_CODE_OVERFLOW;
                     const bool cont_mode        = _last_stream_cmd.at(chan).stream_mode
                                            == stream_cmd_t::STREAM_MODE_START_CONTINUOUS;
-                    _restart_cont[chan]                = cont_mode;
                     rx_event_action->args["cont_mode"] = std::to_string(cont_mode);
                     RFNOC_LOG_TRACE("Posting overrun event action message.");
                     post_action(res_source_info{res_source_info::OUTPUT_EDGE, chan},
diff --git a/host/lib/rfnoc/rfnoc_rx_streamer.cpp b/host/lib/rfnoc/rfnoc_rx_streamer.cpp
index cfa88b292..b50e2fe15 100644
--- a/host/lib/rfnoc/rfnoc_rx_streamer.cpp
+++ b/host/lib/rfnoc/rfnoc_rx_streamer.cpp
@@ -18,16 +18,18 @@ using namespace uhd::rfnoc;
 const std::string STREAMER_ID = "RxStreamer";
 static std::atomic<uint64_t> streamer_inst_ctr;
 
-rfnoc_rx_streamer::rfnoc_rx_streamer(const size_t num_chans,
-    const uhd::stream_args_t stream_args)
+rfnoc_rx_streamer::rfnoc_rx_streamer(
+    const size_t num_chans, const uhd::stream_args_t stream_args)
     : rx_streamer_impl<chdr_rx_data_xport>(num_chans, stream_args)
     , _unique_id(STREAMER_ID + "#" + std::to_string(streamer_inst_ctr++))
     , _stream_args(stream_args)
 {
+    set_overrun_handler([this]() { this->_handle_overrun(); });
+
     // No block to which to forward properties or actions
     set_prop_forwarding_policy(forwarding_policy_t::DROP);
     set_action_forwarding_policy(forwarding_policy_t::DROP);
-    //
+
     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 =
@@ -38,10 +40,6 @@ rfnoc_rx_streamer::rfnoc_rx_streamer(const size_t num_chans,
             }
             _handle_rx_event_action(src, rx_event_action);
         });
-    register_action_handler(ACTION_KEY_RX_RESTART_REQ,
-        [this](const res_source_info& src, action_info::sptr action) {
-            _handle_restart_request(src, action);
-        });
     register_action_handler(ACTION_KEY_STREAM_CMD,
         [this](const res_source_info& src, action_info::sptr action) {
             stream_cmd_action_info::sptr stream_cmd_action =
@@ -117,6 +115,15 @@ bool rfnoc_rx_streamer::check_topology(
     return node_t::check_topology(connected_inputs, connected_outputs);
 }
 
+void rfnoc_rx_streamer::_handle_overrun()
+{
+    if (_overrun_handling_mode) {
+        RFNOC_LOG_TRACE("Requesting restart from overrun-reporting node...");
+        post_action({res_source_info::INPUT_EDGE, _overrun_channel},
+            action_info::make(ACTION_KEY_RX_RESTART_REQ));
+    }
+}
+
 void rfnoc_rx_streamer::_register_props(const size_t chan,
     const std::string& otw_format)
 {
@@ -178,6 +185,7 @@ void rfnoc_rx_streamer::_handle_rx_event_action(
             RFNOC_LOG_TRACE("Ignoring duplicate overrun message.");
             return;
         }
+        _overrun_channel = src.instance;
         RFNOC_LOG_TRACE(
             "Switching to overrun-handling mode: Stopping all upstream producers...");
         auto stop_action =
@@ -188,32 +196,17 @@ void rfnoc_rx_streamer::_handle_rx_event_action(
             post_action({res_source_info::INPUT_EDGE, i}, stop_action);
         }
         if (!rx_event_action->args.cast<bool>("cont_mode", false)) {
-            // FIXME wait until it's safe to restart radios
-            // If we don't need to restart, that's all we need to do
+            // If we don't need to restart, that's all we need to do. Clear this
+            // flag before setting the stopped due to overrun status below to
+            // avoid a potential race condition with the overrun handler.
             _overrun_handling_mode = false;
         }
+        // Tell the streamer to flag an overrun to the user after the data that
+        // was buffered prior to the overrun is read.
+        set_stopped_due_to_overrun();
     }
 }
 
-void rfnoc_rx_streamer::_handle_restart_request(
-    const res_source_info& src, action_info::sptr)
-{
-    // FIXME: Now we need to wait until it's safe to restart the radios.
-    // A flush would achieve this, albeit at the cost of possibly losing
-    // samples.
-    // The earliest we can restart is when the FIFOs in upstream producers
-    // are empty.
-    RFNOC_LOG_TRACE("Waiting for FIFOs to clear");
-
-    std::this_thread::sleep_for(100ms);
-
-    // Once it's safe to restart the radios, we ask a radio to send us a
-    // stream command with its current time.
-    RFNOC_LOG_TRACE("Requesting restart from overrun-reporting node...");
-    post_action({res_source_info::INPUT_EDGE, src.instance},
-        action_info::make(ACTION_KEY_RX_RESTART_REQ));
-}
-
 void rfnoc_rx_streamer::_handle_stream_cmd_action(
     const res_source_info& src, stream_cmd_action_info::sptr stream_cmd_action)
 {