Device3: Improved send flow control

- Implemented zero_copy_flow_ctrl for send transport
- Removed bounded buffer for sequence acks
- Created spin wait on flow control for fastest response to flow control updates

1 files changed, 44 insertions, 24 deletions

diff --git a/host/lib/usrp/device3/device3_io_impl.cpp b/host/lib/usrp/device3/device3_io_impl.cpp
index c1bfd1606..aa84dcab0 100644
--- a/host/lib/usrp/device3/device3_io_impl.cpp
+++ b/host/lib/usrp/device3/device3_io_impl.cpp
@@ -31,6 +31,8 @@
 #include "../../rfnoc/tx_stream_terminator.hpp"
 #include <uhd/rfnoc/rate_node_ctrl.hpp>
 #include <uhd/rfnoc/radio_ctrl.hpp>
+#include <uhd/transport/zero_copy_flow_ctrl.hpp>
+#include <boost/atomic.hpp>
 
 #define UHD_STREAMER_LOG() UHD_LOGV(never)
 
@@ -304,12 +306,13 @@ struct tx_fc_cache_t
         device_channel(0),
         last_seq_out(0),
         last_seq_ack(0),
-        seq_queue(1){}
+        last_seq_ack_cache(0) {}
+
     size_t stream_channel;
     size_t device_channel;
     size_t last_seq_out;
-    size_t last_seq_ack;
-    uhd::transport::bounded_buffer<size_t> seq_queue;
+    boost::atomic_size_t last_seq_ack;
+    size_t last_seq_ack_cache;
     boost::shared_ptr<device3_impl::async_md_type> async_queue;
     boost::shared_ptr<device3_impl::async_md_type> old_async_queue;
 };
@@ -337,33 +340,43 @@ static size_t get_tx_flow_control_window(
     return window_in_pkts;
 }
 
-static managed_send_buffer::sptr get_tx_buff_with_flowctrl(
-    task::sptr /*holds ref*/,
-    boost::shared_ptr<tx_fc_cache_t> fc_cache,
+// TODO: Remove this function
+// This function only exists to make sure the transport is not destroyed
+// until it is no longer needed.
+static managed_send_buffer::sptr get_tx_buff(
     zero_copy_if::sptr xport,
-    size_t fc_window,
     const double timeout
 ){
+    return xport->get_send_buff(timeout);
+}
+
+static bool tx_flow_ctrl(
+    task::sptr /*holds ref*/,
+    boost::shared_ptr<tx_fc_cache_t> fc_cache,
+    size_t fc_window,
+    managed_buffer::sptr
+) {
+    // Busy loop waiting for flow control update.  This is necessary because
+    // at this point there is data trying to be sent and it must be sent as
+    // quickly as possible when the flow control update arrives to avoid
+    // underruns at high rates.  This is also OK because it only occurs when
+    // data needs to be sent and flow control is holding it back.
     while (true)
     {
         // delta is the amount of FC credit we've used up
-        const size_t delta = (fc_cache->last_seq_out & HW_SEQ_NUM_MASK) - (fc_cache->last_seq_ack & HW_SEQ_NUM_MASK);
+        const size_t delta = (fc_cache->last_seq_out & HW_SEQ_NUM_MASK) -
+            (fc_cache->last_seq_ack_cache & HW_SEQ_NUM_MASK);
         // If we want to send another packet, we must have FC credit left
         if ((delta & HW_SEQ_NUM_MASK) < fc_window)
-            break;
-
-        // If credit is all used up, we check seq_queue for more.
-        const bool ok = fc_cache->seq_queue.pop_with_timed_wait(fc_cache->last_seq_ack, timeout);
-        if (not ok) {
-            return managed_send_buffer::sptr(); //timeout waiting for flow control
+        {
+            // Packet will be sent
+            fc_cache->last_seq_out++; //update seq
+            return true;
         }
+        // update the cached value from the atomic
+        fc_cache->last_seq_ack_cache = fc_cache->last_seq_ack;
     }
-
-    managed_send_buffer::sptr buff = xport->get_send_buff(timeout);
-    if (buff) {
-        fc_cache->last_seq_out++; //update seq, this will actually be a send
-    }
-    return buff;
+    return false;
 }
 
 #define DEVICE3_ASYNC_EVENT_CODE_FLOW_CTRL 0
@@ -380,7 +393,9 @@ static void handle_tx_async_msgs(
 ) {
     managed_recv_buffer::sptr buff = xport->get_recv_buff();
     if (not buff)
+    {
         return;
+    }
 
     //extract packet info
     vrt::if_packet_info_t if_packet_info;
@@ -429,8 +444,7 @@ static void handle_tx_async_msgs(
     //The FC response and the burst ack are two indicators that the radio
     //consumed packets. Use them to update the FC metadata
     if (metadata.event_code == DEVICE3_ASYNC_EVENT_CODE_FLOW_CTRL) {
-        const size_t seq = metadata.user_payload[0];
-        fc_cache->seq_queue.push_with_pop_on_full(seq);
+        fc_cache->last_seq_ack = metadata.user_payload[0];
     }
 
     //FC responses don't propagate up to the user so filter them here
@@ -841,13 +855,19 @@ tx_streamer::sptr device3_impl::get_tx_stream(const uhd::stream_args_t &args_)
             }
         }
 
+        // Add flow control
+        xport.send = zero_copy_flow_ctrl::make(
+            xport.send,
+            boost::bind(&tx_flow_ctrl, task, fc_cache, fc_window, _1),
+            NULL);
+
         //Give the streamer a functor to get the send buffer
-        //get_tx_buff_with_flowctrl is static so bind has no lifetime issues
+        //get_tx_buff is static so bind has no lifetime issues
         //xport.send (sptr) is required to add streamer->data-transport lifetime dependency
         //task (sptr) is required to add  a streamer->async-handler lifetime dependency
         my_streamer->set_xport_chan_get_buff(
             stream_i,
-            boost::bind(&get_tx_buff_with_flowctrl, task, fc_cache, xport.send, fc_window, _1)
+            boost::bind(&get_tx_buff, xport.send, _1)
         );
         //Give the streamer a functor handled received async messages
         my_streamer->set_async_receiver(