rfnoc: add rx and tx transports, and amend rfnoc_graph

transports:

Transports build on I/O service and implements flow control and
sequence number checking.

The rx streamer subclass extends the streamer implementation to connect
it to the rfnoc graph. It receives configuration values from property
propagation and configures the streamer accordingly. It also implements
the issue_stream_cmd rx_streamer API method.

Add implementation of rx streamer creation and method to connect it to
an rfnoc block.

rfnoc_graph: Cache more connection info, clarify contract

Summary of changes:
- rfnoc_graph stores more information about static connections at the
  beginning. Some search algorithms are replaced by simpler lookups.
- The contract for connect() was clarified. It is required to call
  connect, even for static connections.

1 files changed, 341 insertions, 0 deletions

diff --git a/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
new file mode 100644
index 000000000..d66e867bc
--- /dev/null
+++ b/host/lib/include/uhdlib/transport/rx_streamer_impl.hpp
@@ -0,0 +1,341 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#ifndef INCLUDED_LIBUHD_RX_STREAMER_IMPL_HPP
+#define INCLUDED_LIBUHD_RX_STREAMER_IMPL_HPP
+
+#include <uhd/config.hpp>
+#include <uhd/convert.hpp>
+#include <uhd/exception.hpp>
+#include <uhd/stream.hpp>
+#include <uhd/types/endianness.hpp>
+#include <uhd/utils/log.hpp>
+#include <uhdlib/transport/rx_streamer_zero_copy.hpp>
+#include <limits>
+#include <vector>
+
+namespace uhd { namespace transport {
+
+namespace detail {
+
+/*!
+ * Cache of metadata for error handling
+ *
+ * If a recv call reads data from multiple packets, and an error occurs in the
+ * second or later packets, recv stops short of the num samps requested and
+ * returns no error. The error is cached for the next call to recv.
+ *
+ * Timeout errors are an exception. Timeouts that occur in the second or later
+ * packets of a recv call stop the recv method but the error is not returned in
+ * the next call. The user can check for this condition since fewer samples are
+ * returned than the number requested.
+ */
+class rx_metadata_cache
+{
+public:
+    //! Stores metadata in the cache, ignoring timeout errors
+    UHD_FORCE_INLINE void store(const rx_metadata_t& metadata)
+    {
+        if (metadata.error_code != rx_metadata_t::ERROR_CODE_TIMEOUT) {
+            _metadata_cache  = metadata;
+            _cached_metadata = true;
+        }
+    }
+
+    //! Checks for cached metadata
+    UHD_FORCE_INLINE bool check(rx_metadata_t& metadata)
+    {
+        if (_cached_metadata) {
+            metadata         = _metadata_cache;
+            _cached_metadata = false;
+            return true;
+        }
+        return false;
+    }
+
+private:
+    // Whether there is a cached metadata object
+    bool _cached_metadata = false;
+
+    // Cached metadata value
+    uhd::rx_metadata_t _metadata_cache;
+};
+
+} // namespace detail
+
+/*!
+ * Implementation of rx streamer API
+ */
+template <typename transport_t>
+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)
+    {
+        if (stream_args.cpu_format.empty()) {
+            throw uhd::value_error("[rx_stream] Must provide a cpu_format!");
+        }
+        if (stream_args.otw_format.empty()) {
+            throw uhd::value_error("[rx_stream] Must provide a otw_format!");
+        }
+        _setup_converters(num_ports, stream_args);
+        _zero_copy_streamer.set_samp_rate(_samp_rate);
+        _zero_copy_streamer.set_bytes_per_item(_convert_info.bytes_per_otw_item);
+    }
+
+    //! Connect a new channel to the streamer
+    void connect_channel(const size_t channel, typename transport_t::uptr xport)
+    {
+        const size_t max_pyld_size = xport->get_max_payload_size();
+        _zero_copy_streamer.connect_channel(channel, std::move(xport));
+
+        // Set spp based on the transport frame size
+        const size_t xport_spp = max_pyld_size / _convert_info.bytes_per_otw_item;
+        _spp = std::min(_spp, xport_spp);
+    }
+
+    //! Implementation of rx_streamer API method
+    size_t get_num_channels() const
+    {
+        return _zero_copy_streamer.get_num_channels();
+    }
+
+    //! Implementation of rx_streamer API method
+    size_t get_max_num_samps() const
+    {
+        return _spp;
+    }
+
+    /*! Get width of each over-the-wire item component. For complex items,
+     *  returns the width of one component only (real or imaginary).
+     */
+    size_t get_otw_item_comp_bit_width() const
+    {
+        return _convert_info.otw_item_bit_width;
+    }
+
+    //! Implementation of rx_streamer API method
+    UHD_INLINE size_t recv(const uhd::rx_streamer::buffs_type& buffs,
+        const size_t nsamps_per_buff,
+        uhd::rx_metadata_t& metadata,
+        const double timeout,
+        const bool one_packet)
+    {
+        if (_error_metadata_cache.check(metadata)) {
+            return 0;
+        }
+
+        const int32_t timeout_ms = static_cast<int32_t>(timeout * 1000);
+
+        size_t total_samps_recv =
+            _recv_one_packet(buffs, nsamps_per_buff, metadata, timeout_ms);
+
+        if (one_packet or metadata.end_of_burst) {
+            return total_samps_recv;
+        }
+
+        // First set of packets recv had an error, return immediately
+        if (metadata.error_code != rx_metadata_t::ERROR_CODE_NONE) {
+            return total_samps_recv;
+        }
+
+        // Loop until buffer is filled or error code. This method returns the
+        // metadata from the first packet received, with the exception of
+        // end-of-burst.
+        uhd::rx_metadata_t loop_metadata;
+
+        while (total_samps_recv < nsamps_per_buff) {
+            size_t num_samps = _recv_one_packet(buffs,
+                nsamps_per_buff - total_samps_recv,
+                loop_metadata,
+                timeout_ms,
+                total_samps_recv * _convert_info.bytes_per_cpu_item);
+
+            // If metadata had an error code set, store for next call and return
+            if (loop_metadata.error_code != rx_metadata_t::ERROR_CODE_NONE) {
+                _error_metadata_cache.store(loop_metadata);
+                break;
+            }
+
+            total_samps_recv += num_samps;
+
+            // Return immediately if end of burst
+            if (loop_metadata.end_of_burst) {
+                metadata.end_of_burst = true;
+                break;
+            }
+        }
+
+        return total_samps_recv;
+    }
+
+protected:
+    //! Configures scaling factor for conversion
+    void set_scale_factor(const size_t chan, const double scale_factor)
+    {
+        _converters[chan]->set_scalar(scale_factor);
+    }
+
+    //! Configures sample rate for conversion of timestamp
+    void set_samp_rate(const double rate)
+    {
+        _samp_rate = rate;
+        _zero_copy_streamer.set_samp_rate(rate);
+    }
+
+    //! Configures tick rate for conversion of timestamp
+    void set_tick_rate(const double rate)
+    {
+        _zero_copy_streamer.set_tick_rate(rate);
+    }
+
+private:
+    //! Converter and associated item sizes
+    struct convert_info
+    {
+        size_t bytes_per_otw_item;
+        size_t bytes_per_cpu_item;
+        size_t otw_item_bit_width;
+    };
+
+    //! Receive a single packet
+    UHD_FORCE_INLINE size_t _recv_one_packet(const uhd::rx_streamer::buffs_type& buffs,
+        const size_t nsamps_per_buff,
+        uhd::rx_metadata_t& metadata,
+        const int32_t timeout_ms,
+        const size_t buffer_offset_bytes = 0)
+    {
+        if (_buff_samps_remaining == 0) {
+            // Current set of buffers has expired, get the next one
+            _buff_samps_remaining =
+                _zero_copy_streamer.get_recv_buffs(_in_buffs, metadata, timeout_ms);
+            _fragment_offset_in_samps = 0;
+        } else {
+            // There are samples still left in the current set of buffers
+            metadata = _last_fragment_metadata;
+            metadata.time_spec += time_spec_t::from_ticks(
+                _fragment_offset_in_samps - metadata.fragment_offset, _samp_rate);
+        }
+
+        if (_buff_samps_remaining != 0) {
+            const size_t num_samps = std::min(nsamps_per_buff, _buff_samps_remaining);
+
+            // Convert samples to the streamer's output format
+            for (size_t i = 0; i < get_num_channels(); i++) {
+                char* b = reinterpret_cast<char*>(buffs[i]);
+                const uhd::rx_streamer::buffs_type out_buffs(b + buffer_offset_bytes);
+                _convert_to_out_buff(out_buffs, i, num_samps);
+            }
+
+            _buff_samps_remaining -= num_samps;
+
+            // Write the fragment flags and offset
+            metadata.more_fragments  = _buff_samps_remaining != 0;
+            metadata.fragment_offset = _fragment_offset_in_samps;
+
+            if (metadata.more_fragments) {
+                _fragment_offset_in_samps += num_samps;
+                _last_fragment_metadata = metadata;
+            }
+
+            return num_samps;
+        } else {
+            return 0;
+        }
+    }
+
+    //! Convert samples for one channel into its buffer
+    UHD_FORCE_INLINE void _convert_to_out_buff(
+        const uhd::rx_streamer::buffs_type& out_buffs,
+        const size_t chan,
+        const size_t num_samps)
+    {
+        const char* buffer_ptr = reinterpret_cast<const char*>(_in_buffs[chan]);
+
+        _converters[chan]->conv(buffer_ptr, out_buffs, num_samps);
+
+        // Advance the pointer for the source buffer
+        _in_buffs[chan] =
+            buffer_ptr + num_samps * _convert_info.bytes_per_otw_item;
+
+        if (_buff_samps_remaining == num_samps) {
+            _zero_copy_streamer.release_recv_buff(chan);
+        }
+    }
+
+    //! Create converters and initialize _convert_info
+    void _setup_converters(const size_t num_ports,
+        const uhd::stream_args_t stream_args)
+    {
+        // Note to code archaeologists: In the past, we had to also specify the
+        // endianness here, but that is no longer necessary because we can make
+        // the wire endianness match the host endianness.
+        convert::id_type id;
+        id.input_format  = stream_args.otw_format + "_chdr";
+        id.num_inputs    = 1;
+        id.output_format = stream_args.cpu_format;
+        id.num_outputs   = 1;
+
+        auto starts_with = [](const std::string& s, const std::string v) {
+            return s.find(v) == 0;
+        };
+
+        const bool otw_is_complex = starts_with(stream_args.otw_format, "fc")
+                                    || starts_with(stream_args.otw_format, "sc");
+
+        convert_info info;
+        info.bytes_per_otw_item = convert::get_bytes_per_item(id.input_format);
+        info.bytes_per_cpu_item = convert::get_bytes_per_item(id.output_format);
+
+        if (otw_is_complex) {
+            info.otw_item_bit_width = info.bytes_per_otw_item * 8 / 2;
+        } else {
+            info.otw_item_bit_width = info.bytes_per_otw_item * 8;
+        }
+
+        _convert_info = info;
+
+        for (size_t i = 0; i < num_ports; i++) {
+            _converters.push_back(convert::get_converter(id)());
+            _converters.back()->set_scalar(1 / 32767.0);
+        }
+    }
+
+    // Converter and item sizes
+    convert_info _convert_info;
+
+    // Converters
+    std::vector<uhd::convert::converter::sptr> _converters;
+
+    // Implementation of frame buffer management and packet info
+    rx_streamer_zero_copy<transport_t> _zero_copy_streamer;
+
+    // Container for buffer pointers used in recv method
+    std::vector<const void*> _in_buffs;
+
+    // Sample rate used to calculate metadata time_spec_t
+    double _samp_rate = 1.0;
+
+    // Maximum number of samples per packet
+    size_t _spp = std::numeric_limits<std::size_t>::max();
+
+    // Num samps remaining in buffer currently held by zero copy streamer
+    size_t _buff_samps_remaining = 0;
+
+    // Metadata cache for error handling
+    detail::rx_metadata_cache _error_metadata_cache;
+
+    // Fragment (partially read packet) information
+    size_t _fragment_offset_in_samps = 0;
+    rx_metadata_t _last_fragment_metadata;
+};
+
+}} // namespace uhd::transport
+
+#endif /* INCLUDED_LIBUHD_RX_STREAMER_IMPL_HPP */