multi_usrp_rfnoc: Add TX buffering using Replay

Enabled with the "tx_replay_buffer" device argument.  Buffers TX data in
DRAM using the Replay block (version 1.1 or higher required), allowing
more buffering of data on the device.  May reduce underruns for certain
applications.  The Replay block is currently limited to 32 play
commands, so fewer calls to send() with larger buffers will perform
better than more calls with smaller buffers.

Signed-off-by: michael-west <michael.west@ettus.com>

6 files changed, 482 insertions, 86 deletions

diff --git a/host/docs/configuration.dox b/host/docs/configuration.dox
index d9a121505..b65714a29 100644
--- a/host/docs/configuration.dox
+++ b/host/docs/configuration.dox
@@ -212,6 +212,10 @@ The following parameters are supported:
 - `noclear` (applies to B100 and N2xx only)
 - `port` and `addr` (N2xx only) These settings specify an alternate receiver
   streamer destination.
+- `streamer` Specify the type of streamer to use.  "replay_buffered" (applies
+  to RFNoC enabled devices with a Replay block in the FPGA image) Adds data
+  buffering in DRAM using the Replay block for TX streamers when using the
+  multi_usrp API.
 
 \subsubsection config_stream_args_transport Transport-related Stream Arguments
 
diff --git a/host/docs/general.dox b/host/docs/general.dox
index 2dab0a9ea..4e1d0cfb9 100644
--- a/host/docs/general.dox
+++ b/host/docs/general.dox
@@ -189,6 +189,18 @@ Underflow occurs when the host does not produce data fast enough. When
 UHD software detects the underflow, it prints a "U" to stdout, and
 pushes a message packet into the async message stream.
 
+Some underruns may be mitigated by buffering data in DRAM using the Replay
+block.  The Replay block is only available in devices that support RFNoC.  Some
+RFNoC devices include the Replay block in the default FPGA image.  Use the
+'uhd_usrp_probe' utility and look for the "RFNoC blocks on this device:"
+section of the output to check if the Replay block is present.  If using
+the multi_usrp API, simply add the stream argument "streamer=replay_buffered"
+to enable the buffering.  This buffering adds latency and will likely not work
+at the highest streaming rates.  A limited number of buffers can be stored
+in the Replay block, so larger buffers supplied to the tx_streamer::send()
+call will produce the best results.  Buffers that are too small will result
+in gaps in the transmitted signal.
+
 <b>Note:</b> "O" and "U" message are generally harmless, and just mean the host
 machine can't keep up with the requested rates.
 
diff --git a/host/lib/include/uhdlib/rfnoc/rfnoc_tx_streamer_replay_buffered.hpp b/host/lib/include/uhdlib/rfnoc/rfnoc_tx_streamer_replay_buffered.hpp
new file mode 100644
index 000000000..23386d8c3
--- /dev/null
+++ b/host/lib/include/uhdlib/rfnoc/rfnoc_tx_streamer_replay_buffered.hpp
@@ -0,0 +1,78 @@
+//
+// Copyright 2022 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#pragma once
+
+#include <uhd/rfnoc_graph.hpp>
+#include <uhd/rfnoc/replay_block_control.hpp>
+#include <uhdlib/rfnoc/rfnoc_tx_streamer.hpp>
+
+namespace uhd { namespace rfnoc {
+
+/*!
+ * Extends the rfnoc_tx_streamer so it can use a Replay block to
+ * buffer TX data.
+ */
+class rfnoc_tx_streamer_replay_buffered : public rfnoc_tx_streamer
+{
+public:
+    struct replay_config_t
+    {
+        replay_block_control::sptr ctrl = nullptr;  // Replay block control
+        size_t port                     = 0;        // Replay port to use
+        uint64_t start_address          = 0;        // Start address in memory
+        uint64_t mem_size               = 0;        // Size of memory block to use
+    };
+
+    struct replay_status_t {
+        const replay_config_t config;
+        uint64_t record_offset          = 0;
+        uint64_t play_offset            = 0;
+        uint64_t play_end               = 0;
+    };
+
+    /*! Constructor
+     *
+     * \param num_ports     The number of ports
+     * \param stream_args   Arguments to aid the construction of the streamer
+     * \param disconnect_cb Callback function to disconnect the streamer when
+     *                      the object is destroyed
+     * \param replay_chans  Vector of Replay configurations to use (one per channel)
+     */
+    rfnoc_tx_streamer_replay_buffered(
+        const size_t num_ports,
+        const uhd::stream_args_t stream_args,
+        std::function<void(const std::string&)> disconnect_cb,
+        std::vector<replay_config_t> replay_configs);
+
+    /*! Destructor
+     */
+    ~rfnoc_tx_streamer_replay_buffered();
+
+    /*! Send
+     * 
+     * Sends data by recording to the Replay block and playing it.
+     *
+     * \param buffs a vector of read-only memory containing samples
+     * \param nsamps_per_buff the number of samples to send, per buffer
+     * \param metadata data describing the buffer's contents
+     * \param timeout the timeout in seconds to wait on a packet
+     * \return the number of samples sent
+     */
+    size_t send(const buffs_type& buffs,
+        const size_t nsamps_per_buff,
+        const tx_metadata_t& metadata,
+        const double timeout = 0.1) override;
+
+private:
+    // Size of item
+    size_t _bytes_per_otw_item;
+
+    // Status of Replay channels
+    std::vector<replay_status_t> _replay_chans;
+};
+
+}} // namespace uhd::rfnoc
diff --git a/host/lib/rfnoc/CMakeLists.txt b/host/lib/rfnoc/CMakeLists.txt
index 53bca9b45..50eb9dd97 100644
--- a/host/lib/rfnoc/CMakeLists.txt
+++ b/host/lib/rfnoc/CMakeLists.txt
@@ -37,6 +37,7 @@ LIBUHD_APPEND_SOURCES(
     ${CMAKE_CURRENT_SOURCE_DIR}/mgmt_portal.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/rfnoc_rx_streamer.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/rfnoc_tx_streamer.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/rfnoc_tx_streamer_replay_buffered.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tx_async_msg_queue.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/mock_block.cpp
     # Default block control classes:
diff --git a/host/lib/rfnoc/rfnoc_tx_streamer_replay_buffered.cpp b/host/lib/rfnoc/rfnoc_tx_streamer_replay_buffered.cpp
new file mode 100644
index 000000000..847d53234
--- /dev/null
+++ b/host/lib/rfnoc/rfnoc_tx_streamer_replay_buffered.cpp
@@ -0,0 +1,181 @@
+//
+// Copyright 2022 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#include <uhdlib/rfnoc/rfnoc_tx_streamer_replay_buffered.hpp>
+#include <uhd/exception.hpp>
+#include <uhd/utils/math.hpp>
+#include <uhd/utils/safe_call.hpp>
+
+using namespace uhd;
+using namespace uhd::rfnoc;
+
+rfnoc_tx_streamer_replay_buffered::rfnoc_tx_streamer_replay_buffered(
+    const size_t num_ports,
+    const uhd::stream_args_t stream_args,
+    std::function<void(const std::string&)> disconnect_cb,
+    std::vector<replay_config_t> replay_configs) :
+    rfnoc_tx_streamer(num_ports, stream_args, disconnect_cb),
+    _bytes_per_otw_item(uhd::convert::get_bytes_per_item(stream_args.otw_format))
+{
+    if(replay_configs.size() != num_ports) {
+        throw uhd::value_error("[TX Streamer] Number of Replay configurations "
+        "does not match the number of channels");
+    }
+
+    for (auto config : replay_configs)
+    {
+        _replay_chans.push_back({config, 0, 0, 0});
+        config.ctrl->set_play_type(stream_args.otw_format);
+    }
+}
+
+rfnoc_tx_streamer_replay_buffered::~rfnoc_tx_streamer_replay_buffered()
+{
+    // Stop all playback
+    for (auto chan : _replay_chans)
+    {
+        UHD_SAFE_CALL(chan.config.ctrl->stop(chan.config.port);)
+    }
+}
+
+size_t rfnoc_tx_streamer_replay_buffered::send(
+        const buffs_type& buffs,
+        const size_t nsamps_per_buff,
+        const tx_metadata_t& metadata,
+        const double timeout)
+{
+    uint64_t record_size = nsamps_per_buff * _bytes_per_otw_item;
+
+    // Make sure there is space in the buffer
+    auto timeout_time = std::chrono::steady_clock::now() +
+                        std::chrono::microseconds(long(timeout * 1000000));
+    for (auto& chan : _replay_chans) {
+        const auto& config = chan.config;
+        const auto& replay = config.ctrl;
+        auto& play_offset = chan.play_offset;
+        const auto& play_end = chan.play_end;
+
+        // Make sure the send does not exceed the memory space
+        if (record_size > config.mem_size) {
+            throw uhd::runtime_error("[multi_usrp] Unable to buffer more than " +
+                std::to_string(config.mem_size) + " bytes");
+        }
+
+        // Make sure nsamps_per_buff is properly aligned to the DRAM
+        if (record_size % replay->get_word_size() != 0) {
+            throw uhd::runtime_error(
+                "[multi_usrp] Number of samples for send() call must be a "
+                "multiple of " + std::to_string(uhd::math::lcm<uint64_t>(
+                    replay->get_word_size(), uint64_t(_bytes_per_otw_item))) +
+                    " for DRAM alignment");
+        }
+
+        // The buffer can be full or empty when play_offset and play_end
+        // are the same, so subtract one from the calculated room to make
+        // sure the buffer is never absolutely full and it can be assumed
+        // the buffer is empty when they are the same.
+        auto room = (play_end == play_offset ? config.mem_size - 1 :
+            play_end < play_offset ? play_offset - play_end - 1 :
+            std::max<uint64_t>(config.mem_size - play_end - 1,
+            play_offset - 1));
+        while (room < record_size) {
+            // Return 0 if timeout
+            if (std::chrono::steady_clock::now() > timeout_time) {
+                UHD_LOG_TRACE("MULTI_USRP",
+                    "send() timed out waiting for room in buffer");
+                return 0;
+            }
+            try {
+                play_offset = replay->get_play_position(config.port) -
+                    config.start_address;
+            } catch (uhd::op_timeout&) {
+                // Internal timeout trying to read the register
+                continue;
+            }
+            room = (play_end == play_offset ? config.mem_size - 1 :
+                play_end < play_offset ? play_offset - play_end - 1 :
+                std::max<uint64_t>(config.mem_size - play_end - 1,
+                play_offset - 1));
+        }
+    }
+
+    // Set up replay blocks to record
+    for (auto& chan : _replay_chans) {
+        const auto& config = chan.config;
+        const auto& replay = config.ctrl;
+        const auto& play_end = chan.play_end;
+        auto& record_offset = chan.record_offset;
+
+        // Default to using space at end of last playback
+        record_offset = play_end;
+        // Change to beginning of memory block if not enough room at end
+        if (config.mem_size - record_offset < record_size) {
+            record_offset = 0;
+        }
+        while (1) {
+            try {
+                replay->record(config.start_address + record_offset, record_size,
+                    config.port);
+                break;
+            } catch(uhd::op_timeout& e) {
+                // Internal timeout trying to write the registers
+                // Return 0 if timeout
+                if (std::chrono::steady_clock::now() > timeout_time) {
+                    UHD_LOG_TRACE("MULTI_USRP",
+                        std::string("send() timed out while setting up to record: ") +
+                        e.what());
+                    return 0;
+                }
+            }
+        }
+    }
+
+    // Send data to replay blocks
+    auto num_samps = rfnoc_tx_streamer::send(
+        buffs, nsamps_per_buff, metadata, timeout);
+
+    if (num_samps) {
+        // Play data
+        for (auto& chan : _replay_chans) {
+            const auto& config = chan.config;
+            const auto& replay = config.ctrl;
+            const auto& record_offset = chan.record_offset;
+            const uint64_t play_start = config.start_address + record_offset;
+            const uint64_t play_size = num_samps * _bytes_per_otw_item;
+            auto& play_end = chan.play_end;
+
+            while (1) {
+                try {
+                    replay->play(play_start, play_size, config.port,
+                        metadata.has_time_spec ? metadata.time_spec :
+                        uhd::time_spec_t(0.0));
+                    break;
+                } catch(uhd::op_failed& e) {
+                    // Too many play commands in queue 
+                    // Return 0 if timeout.
+                    if (std::chrono::steady_clock::now() > timeout_time) {
+                        UHD_LOG_TRACE("MULTI_USRP",
+                            std::string("send() timed out issuing play command: ") +
+                            e.what());
+                        return 0;
+                    }
+                } catch (uhd::op_timeout& e) {
+                    // Internal timeout trying to write the registers
+                    // Return 0 if timeout
+                    if (std::chrono::steady_clock::now() > timeout_time) {
+                        UHD_LOG_TRACE("MULTI_USRP",
+                            std::string("send() timed out issuing play command: ") +
+                            e.what());
+                        return 0;
+                    }
+                }
+            }
+
+            play_end = record_offset + play_size;
+        }
+    }
+    return num_samps;
+}
\ No newline at end of file
diff --git a/host/lib/usrp/multi_usrp_rfnoc.cpp b/host/lib/usrp/multi_usrp_rfnoc.cpp
index c71668c33..00e386b71 100644
--- a/host/lib/usrp/multi_usrp_rfnoc.cpp
+++ b/host/lib/usrp/multi_usrp_rfnoc.cpp
@@ -1,22 +1,26 @@
 //
-// Copyright 2019 Ettus Research, a National Instruments Brand
+// Copyright 2019-2022 Ettus Research, a National Instruments Brand
 //
 // SPDX-License-Identifier: GPL-3.0-or-later
 //
 
+#include <uhd/convert.hpp>
 #include <uhd/exception.hpp>
 #include <uhd/rfnoc/ddc_block_control.hpp>
 #include <uhd/rfnoc/duc_block_control.hpp>
 #include <uhd/rfnoc/filter_node.hpp>
 #include <uhd/rfnoc/graph_edge.hpp>
 #include <uhd/rfnoc/radio_control.hpp>
+#include <uhd/rfnoc/replay_block_control.hpp>
 #include <uhd/rfnoc_graph.hpp>
 #include <uhd/types/device_addr.hpp>
 #include <uhd/usrp/multi_usrp.hpp>
 #include <uhd/utils/graph_utils.hpp>
+#include <uhd/utils/math.hpp>
 #include <uhdlib/rfnoc/rfnoc_device.hpp>
 #include <uhdlib/rfnoc/rfnoc_rx_streamer.hpp>
 #include <uhdlib/rfnoc/rfnoc_tx_streamer.hpp>
+#include <uhdlib/rfnoc/rfnoc_tx_streamer_replay_buffered.hpp>
 #include <uhdlib/usrp/gpio_defs.hpp>
 #include <uhdlib/usrp/multi_usrp_utils.hpp>
 #include <uhdlib/utils/narrow.hpp>
@@ -157,6 +161,8 @@ std::string bytes_to_str(std::vector<uint8_t> str_b)
 
 class multi_usrp_rfnoc : public multi_usrp
 {
+    using replay_config_t = rfnoc_tx_streamer_replay_buffered::replay_config_t;
+
 public:
     struct rx_chan_t
     {
@@ -172,6 +178,7 @@ public:
         duc_block_control::sptr duc; // can be nullptr
         size_t block_chan;
         std::vector<graph_edge_t> edge_list;
+        replay_config_t replay;
     };
 
     /**************************************************************************
@@ -184,60 +191,62 @@ public:
         , _device(std::make_shared<redirector_device>(this))
     {
         // Discover all of the radios on our devices and create a mapping between
-        // radio chains and channel numbers. The result is sorted.
-        auto radio_blk_ids = _graph->find_blocks("Radio");
-        // If we don't find any radios, we don't have a multi_usrp object
-        if (radio_blk_ids.empty()) {
-            throw uhd::runtime_error(
-                "[multi_usrp] No radios found in connected devices.");
-        }
-        // Next, we assign block controllers to RX channels
-        // Note that we don't want to connect blocks now; we will wait until we create and
-        // connect a streamer. This gives us a little more time to figure out the desired
-        // values of our properties (such as master clock)
+        // radio chains and channel numbers.  Do this one motherboard at a time
+        // because each device can have a different subdev spec.
         size_t musrp_rx_channel = 0;
         size_t musrp_tx_channel = 0;
-        for (auto radio_id : radio_blk_ids) {
-            auto radio_blk = _graph->get_block<uhd::rfnoc::radio_control>(radio_id);
-
-            // Store radio blocks per mboard for quick retrieval
-            _radios[radio_id.get_device_no()].push_back(radio_blk);
-
-            for (size_t block_chan = 0; block_chan < radio_blk->get_num_output_ports();
-                 ++block_chan) {
-                // Create the RX chan
-                uhd::usrp::subdev_spec_t rx_radio_subdev;
-                rx_radio_subdev.push_back(uhd::usrp::subdev_spec_pair_t(
-                    radio_blk->get_slot_name(),
-                    radio_blk->get_dboard_fe_from_chan(block_chan, uhd::RX_DIRECTION)));
-                auto rx_chans =
-                    _generate_mboard_rx_chans(rx_radio_subdev, radio_id.get_device_no());
-                // TODO: we're passing the same info around here; there has to be a
-                // cleaner way
-                for (auto rx_chan : rx_chans) {
-                    _rx_chans.emplace(musrp_rx_channel, rx_chan);
-                    ++musrp_rx_channel; // Increment after logging so we print the correct
-                                        // value
+        for (size_t mboard = 0; mboard < get_num_mboards(); mboard++)
+        {
+            auto radio_blk_ids = _graph->find_blocks(std::to_string(mboard) + "/Radio");
+
+            // Generate the RX and TX subdev specs
+            uhd::usrp::subdev_spec_t rx_radio_subdev;
+            uhd::usrp::subdev_spec_t tx_radio_subdev;
+            for (auto radio_id : radio_blk_ids) {
+                auto radio_blk = _graph->get_block<uhd::rfnoc::radio_control>(radio_id);
+
+                // Store radio blocks per mboard for quick retrieval
+                _radios[mboard].push_back(radio_blk);
+
+                for (size_t block_chan = 0; block_chan < radio_blk->get_num_output_ports();
+                    ++block_chan) {
+                    rx_radio_subdev.push_back(uhd::usrp::subdev_spec_pair_t(
+                        radio_blk->get_slot_name(),
+                        radio_blk->get_dboard_fe_from_chan(block_chan, uhd::RX_DIRECTION)));
                 }
-            }
-            for (size_t block_chan = 0; block_chan < radio_blk->get_num_input_ports();
-                 ++block_chan) {
-                // Create the TX chan
-                uhd::usrp::subdev_spec_t tx_radio_subdev;
-                tx_radio_subdev.push_back(uhd::usrp::subdev_spec_pair_t(
-                    radio_blk->get_slot_name(),
-                    radio_blk->get_dboard_fe_from_chan(block_chan, uhd::TX_DIRECTION)));
-                auto tx_chans =
-                    _generate_mboard_tx_chans(tx_radio_subdev, radio_id.get_device_no());
-                // TODO: we're passing the same info around here; there has to be a
-                // cleaner way
-                for (auto tx_chan : tx_chans) {
-                    _tx_chans.emplace(musrp_tx_channel, tx_chan);
-                    ++musrp_tx_channel; // Increment after logging so we print the correct
-                                        // value
+
+                for (size_t block_chan = 0; block_chan < radio_blk->get_num_input_ports();
+                    ++block_chan) {
+                    tx_radio_subdev.push_back(uhd::usrp::subdev_spec_pair_t(
+                        radio_blk->get_slot_name(),
+                        radio_blk->get_dboard_fe_from_chan(block_chan, uhd::TX_DIRECTION)));
                 }
             }
+
+            // Generate the TX and RX chans
+            // Note that we don't want to connect blocks now; we will wait until we create and
+            // connect a streamer. This gives us a little more time to figure out the desired
+            // values of our properties (such as master clock)
+            auto rx_chans =
+                _generate_mboard_rx_chans(rx_radio_subdev, mboard);
+            for (auto rx_chan : rx_chans) {
+                _rx_chans.emplace(musrp_rx_channel, rx_chan);
+                ++musrp_rx_channel;
+            }
+            auto tx_chans =
+                _generate_mboard_tx_chans(tx_radio_subdev, mboard);
+            for (auto tx_chan : tx_chans) {
+                _tx_chans.emplace(musrp_tx_channel, tx_chan);
+                ++musrp_tx_channel;
+            }
         }
+
+        if (_rx_chans.empty() and _tx_chans.empty()) {
+            // If we don't find any valid radio chains, we don't have a multi_usrp object
+            throw uhd::runtime_error(
+                "[multi_usrp] No valid radio channels found in connected devices.");
+        }
+
         // Manually propagate radio block sample rates to DDC/DUC blocks in order to allow
         // DDC/DUC blocks to have valid internal state before graph is (later) connected
         for (size_t rx_chan = 0; rx_chan < get_rx_num_channels(); ++rx_chan) {
@@ -254,6 +263,7 @@ public:
                     tx_chain.radio->get_rate(), tx_chain.block_chan);
             }
         }
+
         _graph->commit();
     }
 
@@ -362,6 +372,8 @@ public:
         std::lock_guard<std::recursive_mutex> l(_graph_mutex);
         stream_args_t args = sanitize_stream_args(args_);
         double rate        = 1.0;
+        bool replay_buffered = (args.args.has_key("streamer") and
+                            args.args["streamer"] == "replay_buffered");
 
         // Note that we don't release the graph, which means that property
         // propagation is possible. This is necessary so we don't disrupt
@@ -369,41 +381,63 @@ public:
         // property propagation where possible.
 
         // Connect the chains
-        auto edges = _connect_tx_chains(args.channels);
-        std::weak_ptr<rfnoc_graph> graph_ref(_graph);
+        std::map< size_t, std::vector<graph_edge_t> > edge_lists;
+        std::vector<replay_config_t> replay_configs;
+        for (auto channel : args.channels) {
+            if (replay_buffered) {
+                edge_lists[channel] = _connect_tx_chain_with_replay(channel);
+                replay_configs.push_back(_get_tx_chan(channel).replay);
+            } else {
+                edge_lists[channel] = _connect_tx_chain(channel);
+            }
+        }
 
         // Create a streamer
         // The disconnect callback must disconnect the entire chain because the radio
         // relies on the connections to determine what is enabled.
-        auto tx_streamer = std::make_shared<rfnoc_tx_streamer>(
-            args.channels.size(), args, [=](const std::string& id) {
+        tx_streamer::sptr tx_streamer;
+        std::weak_ptr<rfnoc_graph> graph_ref(_graph);
+        auto disconnect = [=](const std::string& id) {
                 if (auto graph = graph_ref.lock()) {
                     graph->disconnect(id);
-                    for (auto edge : edges) {
-                        graph->disconnect(edge.src_blockid,
-                            edge.src_port,
-                            edge.dst_blockid,
-                            edge.dst_port);
+                    for (auto edge_list : edge_lists) {
+                        for (auto edge : edge_list.second) {
+                            if (block_id_t(edge.src_blockid).match(NODE_ID_SEP) or
+                                block_id_t(edge.dst_blockid).match(NODE_ID_SEP)) {
+                                    continue;
+                            }
+                            graph->disconnect(edge.src_blockid,
+                                edge.src_port,
+                                edge.dst_blockid,
+                                edge.dst_port);
+                        }
                     }
                 }
-            });
+            };
+        if (replay_buffered) {
+            tx_streamer = std::make_shared<rfnoc_tx_streamer_replay_buffered>(
+                    args.channels.size(), args, disconnect, replay_configs);
+        } else {
+            tx_streamer = std::make_shared<rfnoc_tx_streamer>(
+                args.channels.size(), args, disconnect);
+        }
 
         // Connect the streamer
         for (size_t strm_port = 0; strm_port < args.channels.size(); ++strm_port) {
             auto tx_channel = args.channels.at(strm_port);
-            auto tx_chain   = _get_tx_chan(tx_channel);
-            if (tx_chain.edge_list.empty()) {
+            auto edge_list  = edge_lists[tx_channel];
+            if (edge_list.empty()) {
                 throw uhd::runtime_error("Graph edge list is empty for tx channel "
                                          + std::to_string(tx_channel));
             }
             UHD_LOG_TRACE("MULTI_USRP",
                 "Connecting TxStreamer:" << strm_port << " -> "
-                                         << tx_chain.edge_list.back().dst_blockid << ":"
-                                         << tx_chain.edge_list.back().dst_port);
+                                         << edge_list.back().dst_blockid << ":"
+                                         << edge_list.back().dst_port);
             _graph->connect(tx_streamer,
                 strm_port,
-                tx_chain.edge_list.back().dst_blockid,
-                tx_chain.edge_list.back().dst_port);
+                edge_list.back().dst_blockid,
+                edge_list.back().dst_port);
             const double chan_rate =
                 _tx_rates.count(tx_channel) ? _tx_rates.at(tx_channel) : 1.0;
             if (chan_rate > 1.0 && rate != chan_rate) {
@@ -1585,15 +1619,15 @@ public:
                                    "Disabling DUC control.";
                             break;
                         }
-                        auto ddc_blk = _graph->get_block<uhd::rfnoc::duc_block_control>(
+                        auto duc_blk = _graph->get_block<uhd::rfnoc::duc_block_control>(
                             edge.src_blockid);
                         return std::tuple<uhd::rfnoc::duc_block_control::sptr, size_t>(
-                            ddc_blk, block_chan);
+                            duc_blk, block_chan);
                     }
                 }
             } catch (const uhd::exception&) {
                 UHD_LOGGER_DEBUG("MULTI_USRP")
-                    << "No DDC found for radio block " << radio_id << ":"
+                    << "No DUC found for radio block " << radio_id << ":"
                     << std::to_string(block_chan);
                 // Then just return a nullptr
             }
@@ -1602,13 +1636,14 @@ public:
 
         // Create the TX chan
         return tx_chan_t(
-            {radio_blk, std::get<0>(duc_port_def), block_chan, radio_sink_chain});
+            {radio_blk, std::get<0>(duc_port_def), block_chan, radio_sink_chain, replay_config_t()});
     }
 
+    // Generate the TX chains.  Must call with a complete subdev spec for the mboard.
     std::vector<tx_chan_t> _generate_mboard_tx_chans(
         const uhd::usrp::subdev_spec_t& spec, size_t mboard)
     {
-        // Discover all of the radios on our devices and create a mapping between radio
+        // Discover all of the radios on the device and create a mapping between radio
         // chains and channel numbers
         auto radio_blk_ids = _graph->find_blocks(std::to_string(mboard) + "/Radio");
         // If we don't find any radios, we don't have a multi_usrp object
@@ -1618,6 +1653,19 @@ public:
                 + std::to_string(mboard));
         }
 
+        // Set up to map Replay blocks and ports to channel in case the user
+        // requests Replay buffering on the TX streamer
+        const auto replay_blk_ids = _graph->find_blocks(std::to_string(mboard) + "/Replay");
+        size_t replay_index = 0;
+        size_t replay_port = 0;
+        size_t num_replay_ports = 0;
+        for (const auto& replay_id : replay_blk_ids) {
+            auto replay = _graph->get_block<uhd::rfnoc::replay_block_control>(replay_id);
+            num_replay_ports += std::min(
+                replay->get_num_input_ports(),
+                replay->get_num_output_ports());
+        }
+
         // Iterate through the subdev pairs, and try to find a radio that matches
         std::vector<tx_chan_t> new_chans;
         for (auto chan_subdev_pair : spec) {
@@ -1636,8 +1684,43 @@ public:
                 }
                 subdev_spec_pair_t radio_subdev(radio_blk->get_slot_name(),
                     radio_blk->get_dboard_fe_from_chan(block_chan, uhd::TX_DIRECTION));
+
                 if (chan_subdev_pair == radio_subdev) {
-                    new_chans.push_back(_generate_tx_radio_chan(radio_id, block_chan));
+                    tx_chan_t tx_chan(_generate_tx_radio_chan(radio_id, block_chan));
+
+                    // Map a Replay block and port to the channel if there are
+                    // enough Replay ports to cover all the channels
+                    if (num_replay_ports >= spec.size()) {
+                        auto replay_id = replay_blk_ids[replay_index];
+                        auto replay = _graph->get_block<uhd::rfnoc::replay_block_control>(replay_id);
+                        size_t num_ports = std::min(replay->get_num_input_ports(),
+                                                    replay->get_num_output_ports());
+                        while (replay_port >= num_ports) {
+                            // All ports on the current Replay block are allocated.
+                            // Get the next Replay block
+                            replay_index++;
+                            replay_port = 0;
+                            replay_id = replay_blk_ids[replay_index];
+                            replay = _graph->get_block<uhd::rfnoc::replay_block_control>(replay_id);
+                            num_ports = std::min(replay->get_num_input_ports(),
+                                                replay->get_num_output_ports());
+                        }
+
+                        // Update the Replay configuration (including memory allocation)
+                        const auto mem_per_block = replay->get_mem_size() / replay_blk_ids.size();
+                        tx_chan.replay.ctrl = replay;
+                        tx_chan.replay.port = replay_port;
+                        tx_chan.replay.mem_size = mem_per_block / num_ports;
+                        tx_chan.replay.start_address = (replay_index * mem_per_block) +
+                                                (tx_chan.replay.mem_size * replay_port);
+
+                        // Get the next Replay port
+                        replay_port++;
+                    } else {
+                        tx_chan.replay.ctrl == nullptr;
+                    }
+
+                    new_chans.push_back(tx_chan);
                     subdev_found = true;
                 }
             }
@@ -1673,7 +1756,7 @@ public:
     {
         uhd::usrp::subdev_spec_t result;
         for (size_t tx_chan = 0; tx_chan < get_tx_num_channels(); tx_chan++) {
-            auto& tx_chain = _tx_chans.at(tx_chan);
+            auto& tx_chain = _get_tx_chan(tx_chan);
             if (tx_chain.radio->get_block_id().get_device_no() == mboard) {
                 result.push_back(
                     uhd::usrp::subdev_spec_pair_t(tx_chain.radio->get_slot_name(),
@@ -2403,24 +2486,61 @@ private:
         return edges;
     }
 
-    std::vector<graph_edge_t> _connect_tx_chains(std::vector<size_t> chans)
+    std::vector<graph_edge_t> _connect_tx_chain(const size_t chan)
     {
         std::vector<graph_edge_t> edges;
-        for (auto chan : chans) {
-            UHD_LOG_TRACE(
-                "MULTI_USRP", std::string("Connecting TX chain for channel ") + std::to_string(chan));
-            auto chain = _tx_chans.at(chan);
-            for (auto edge : chain.edge_list) {
-                if (block_id_t(edge.src_blockid).match(NODE_ID_SEP)) {
-                    break;
-                }
-                UHD_LOG_TRACE(
-                    "MULTI_USRP", std::string("Connecting TX edge: ") + edge.to_string());
-                _graph->connect(
-                    edge.src_blockid, edge.src_port, edge.dst_blockid, edge.dst_port);
-                edges.push_back(edge);
+
+        UHD_LOG_TRACE(
+            "MULTI_USRP", std::string("Connecting TX chain for channel ") + std::to_string(chan));
+        auto chain = _get_tx_chan(chan);
+        for (auto edge : chain.edge_list) {
+            edges.push_back(edge);
+            if (block_id_t(edge.src_blockid).match(NODE_ID_SEP)) {
+                break;
             }
+            UHD_LOG_TRACE(
+                "MULTI_USRP", std::string("Connecting TX edge: ") + edge.to_string());
+            _graph->connect(
+                edge.src_blockid, edge.src_port, edge.dst_blockid, edge.dst_port);
         }
+
+        return edges;
+    }
+
+    std::vector<graph_edge_t> _connect_tx_chain_with_replay(size_t chan)
+    {
+        std::vector<graph_edge_t> edges;
+        auto tx_chan = _get_tx_chan(chan);
+        if (not tx_chan.replay.ctrl) {
+            throw uhd::runtime_error(
+                "[multi_usrp] No Replay block found to buffer TX stream");
+        }
+        auto replay_id = tx_chan.replay.ctrl->get_block_id();
+        auto replay_port = tx_chan.replay.port;
+
+        // Connect Replay out to Radio in
+        try {
+            edges = connect_through_blocks(
+                _graph,
+                replay_id,
+                replay_port,
+                tx_chan.radio->get_block_id(),
+                tx_chan.block_chan);
+        } catch (uhd::runtime_error& e) {
+            throw uhd::runtime_error(
+                std::string("[multi_usrp] Unable to connect Replay block to Radio: ") +
+                e.what());
+        }
+
+        // Add Replay input edge (for streamer connection)
+        auto replay_edges_in = get_block_chain(_graph, replay_id, replay_port, false);
+        if (replay_edges_in.size() > 1) {
+            throw uhd::runtime_error(
+                "[multi_usrp] Unable to connect TX streamer to Replay block: "
+                "Unexpected block found before Replay block");
+        }
+        edges.push_back(replay_edges_in.at(0));
+
         return edges;
     }