rfnoc: transport: Check if streamers are connected in send() and recv()

This adds a check in send() and recv() whether or not the streamer is
actually connected. If not, an exception is thrown with the message:

    [rx_stream] Attempting to call recv() before all channels are connected!

or

    [tx_stream] Attempting to call send() before all channels are connected!

The check is a single boolean flag check, but it does add a branch in
our hot code. Since this event is unlikely, and only happens in badly
configured apps, we will get some help from the CPUs branch prediction
to reduce the additional cost of this check.

1 files changed, 19 insertions, 1 deletions

diff --git a/host/lib/include/uhdlib/transport/tx_streamer_impl.hpp b/host/lib/include/uhdlib/transport/tx_streamer_impl.hpp
index 9dc3b0c35..6b34c1c10 100644
--- a/host/lib/include/uhdlib/transport/tx_streamer_impl.hpp
+++ b/host/lib/include/uhdlib/transport/tx_streamer_impl.hpp
@@ -10,9 +10,10 @@
 #include <uhd/convert.hpp>
 #include <uhd/stream.hpp>
 #include <uhd/types/metadata.hpp>
-#include <uhd/utils/tasks.hpp>
 #include <uhd/utils/log.hpp>
+#include <uhd/utils/tasks.hpp>
 #include <uhdlib/transport/tx_streamer_zero_copy.hpp>
+#include <algorithm>
 #include <limits>
 #include <vector>
 
@@ -106,6 +107,7 @@ public:
         : _zero_copy_streamer(num_chans)
         , _zero_buffs(num_chans, &_zero)
         , _out_buffs(num_chans)
+        , _chans_connected(num_chans, false)
     {
         _setup_converters(num_chans, stream_args);
         _zero_copy_streamer.set_bytes_per_item(_convert_info.bytes_per_otw_item);
@@ -120,6 +122,11 @@ public:
         const size_t mtu = xport->get_mtu();
         _hdr_len = std::max(_hdr_len, xport->get_chdr_hdr_len());
         _zero_copy_streamer.connect_channel(channel, std::move(xport));
+        // Note: The previous call also checks if the channel index was valid.
+        _chans_connected[channel] = true;
+        _all_chans_connected      = std::all_of(_chans_connected.cbegin(),
+            _chans_connected.cend(),
+            [](const bool connected) { return connected; });
 
         if (mtu < _mtu) {
             set_mtu(mtu);
@@ -149,6 +156,10 @@ public:
         const uhd::tx_metadata_t& metadata_,
         const double timeout) override
     {
+        if (!_all_chans_connected) {
+            throw uhd::runtime_error("[tx_stream] Attempting to call send() before all "
+                                     "channels are connected!");
+        }
         uhd::tx_metadata_t metadata(metadata_);
 
         if (nsamps_per_buff == 0 && metadata.start_of_burst) {
@@ -459,6 +470,13 @@ private:
 
     // Metadata cache for send calls with no data
     detail::tx_metadata_cache _metadata_cache;
+
+    // Store a list of channels that are already connected
+    std::vector<bool> _chans_connected;
+
+    // Flag to store if all channels are connected. This is to speed up the lookup
+    // of all channels' connected-status.
+    bool _all_chans_connected = false;
 };
 
 }} // namespace uhd::transport