2 files changed, 40 insertions, 2 deletions
diff --git a/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
index f776d9373..1b25b308b 100644
--- a/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
+++ b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
@@ -13,6 +13,7 @@
 #include <uhd/types/endianness.hpp>
 #include <uhd/utils/log.hpp>
 #include <uhdlib/transport/rx_streamer_zero_copy.hpp>
+#include <algorithm>
 #include <limits>
 #include <vector>
 
@@ -74,7 +75,9 @@ class rx_streamer_impl : public rx_streamer
 public:
     //! Constructor
     rx_streamer_impl(const size_t num_ports, const uhd::stream_args_t stream_args)
-        : _zero_copy_streamer(num_ports), _in_buffs(num_ports)
+        : _zero_copy_streamer(num_ports)
+        , _in_buffs(num_ports)
+        , _chans_connected(num_ports, false)
     {
         if (stream_args.cpu_format.empty()) {
             throw uhd::value_error("[rx_stream] Must provide a cpu_format!");
@@ -99,6 +102,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);
@@ -132,6 +140,11 @@ public:
         const double timeout,
         const bool one_packet) override
     {
+        if (!_all_chans_connected) {
+            throw uhd::runtime_error("[rx_stream] Attempting to call recv() before all "
+                                     "channels are connected!");
+        }
+
         if (_error_metadata_cache.check(metadata)) {
             return 0;
         }
@@ -422,6 +435,13 @@ private:
     // Fragment (partially read packet) information
     size_t _fragment_offset_in_samps = 0;
     rx_metadata_t _last_fragment_metadata;
+
+    // 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
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