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.

8 files changed, 627 insertions, 290 deletions

diff --git a/host/lib/rfnoc/CMakeLists.txt b/host/lib/rfnoc/CMakeLists.txt
index dfef4f90f..963458fe6 100644
--- a/host/lib/rfnoc/CMakeLists.txt
+++ b/host/lib/rfnoc/CMakeLists.txt
@@ -53,6 +53,8 @@ LIBUHD_APPEND_SOURCES(
     ${CMAKE_CURRENT_SOURCE_DIR}/tick_node_ctrl.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/tx_stream_terminator.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/wb_iface_adapter.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/rfnoc_rx_streamer.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/rfnoc_tx_streamer.cpp
     # Default block control classes:
     ${CMAKE_CURRENT_SOURCE_DIR}/block_control.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/ddc_block_control.cpp
diff --git a/host/lib/rfnoc/chdr_packet.cpp b/host/lib/rfnoc/chdr_packet.cpp
index 653181c04..d4b7494e2 100644
--- a/host/lib/rfnoc/chdr_packet.cpp
+++ b/host/lib/rfnoc/chdr_packet.cpp
@@ -31,7 +31,7 @@ public:
     {
         assert(pkt_buff);
         _pkt_buff = const_cast<uint64_t*>(reinterpret_cast<const uint64_t*>(pkt_buff));
-        _compute_mdata_offset();
+        _mdata_offset = _compute_mdata_offset(get_chdr_header());
     }
 
     virtual void refresh(void* pkt_buff, chdr_header& header, uint64_t timestamp = 0)
@@ -42,7 +42,7 @@ public:
         if (_has_timestamp(header)) {
             _pkt_buff[1] = u64_from_host(timestamp);
         }
-        _compute_mdata_offset();
+        _mdata_offset = _compute_mdata_offset(get_chdr_header());
     }
 
     virtual void update_payload_size(size_t payload_size_bytes)
@@ -115,19 +115,27 @@ public:
             + (chdr_w_stride * (_mdata_offset + get_chdr_header().get_num_mdata())));
     }
 
+    virtual size_t calculate_payload_offset(const packet_type_t pkt_type,
+        const uint8_t num_mdata = 0) const
+    {
+        chdr_header header;
+        header.set_pkt_type(pkt_type);
+        return (_compute_mdata_offset(header) + num_mdata) * chdr_w_bytes;
+    }
+
 private:
     inline bool _has_timestamp(const chdr_header& header) const
     {
         return (header.get_pkt_type() == PKT_TYPE_DATA_WITH_TS);
     }
 
-    inline void _compute_mdata_offset() const
+    inline size_t _compute_mdata_offset(const chdr_header& header) const
     {
         // The metadata offset depends on the chdr_w and whether we have a timestamp
         if (chdr_w == 64) {
-            _mdata_offset = _has_timestamp(get_chdr_header()) ? 2 : 1;
+            return _has_timestamp(header) ? 2 : 1;
         } else {
-            _mdata_offset = 1;
+            return 1;
         }
     }
 
diff --git a/host/lib/rfnoc/graph_stream_manager.cpp b/host/lib/rfnoc/graph_stream_manager.cpp
index f2024786a..2db68db04 100644
--- a/host/lib/rfnoc/graph_stream_manager.cpp
+++ b/host/lib/rfnoc/graph_stream_manager.cpp
@@ -8,7 +8,9 @@
 #include <uhd/utils/log.hpp>
 #include <uhdlib/rfnoc/graph_stream_manager.hpp>
 #include <uhdlib/rfnoc/link_stream_manager.hpp>
+#include <uhdlib/rfnoc/chdr_rx_data_xport.hpp>
 #include <boost/format.hpp>
+#include <boost/make_shared.hpp>
 #include <map>
 #include <set>
 
@@ -175,6 +177,38 @@ public:
                                  "specified source endpoint");
     }
 
+    chdr_rx_data_xport::uptr create_device_to_host_data_stream(
+        const sep_addr_t src_addr,
+        const sw_buff_t pyld_buff_fmt,
+        const sw_buff_t mdata_buff_fmt,
+        const device_addr_t& xport_args)
+    {
+        // TODO: choose a route
+        const device_id_t via_device = NULL_DEVICE_ID;
+
+        return _link_mgrs.at(_check_dst_and_find_src(src_addr, via_device))
+            ->create_device_to_host_data_stream(src_addr,
+                pyld_buff_fmt,
+                mdata_buff_fmt,
+                xport_args);
+    }
+
+    virtual chdr_tx_data_xport::uptr create_host_to_device_data_stream(
+        sep_addr_t dst_addr,
+        const sw_buff_t pyld_buff_fmt,
+        const sw_buff_t mdata_buff_fmt,
+        const device_addr_t& xport_args)
+    {
+        // TODO: choose a route
+        const device_id_t via_device = NULL_DEVICE_ID;
+
+        return _link_mgrs.at(_check_dst_and_find_src(dst_addr, via_device))
+            ->create_host_to_device_data_stream(dst_addr,
+                pyld_buff_fmt,
+                mdata_buff_fmt,
+                xport_args);
+    }
+
 private:
     device_id_t _check_dst_and_find_src(sep_addr_t dst_addr, device_id_t via_device) const
     {
diff --git a/host/lib/rfnoc/link_stream_manager.cpp b/host/lib/rfnoc/link_stream_manager.cpp
index 4fe183529..6855162de 100644
--- a/host/lib/rfnoc/link_stream_manager.cpp
+++ b/host/lib/rfnoc/link_stream_manager.cpp
@@ -203,86 +203,71 @@ public:
                 false,
                 STREAM_SETUP_TIMEOUT);
 
-        // Compute FC frequency and headroom based on buff parameters
-        stream_buff_params_t fc_freq{
-            static_cast<uint64_t>(std::ceil(double(buff_params.bytes) * fc_freq_ratio)),
-            static_cast<uint32_t>(
-                std::ceil(double(buff_params.packets) * fc_freq_ratio))};
-        stream_buff_params_t fc_headroom{
-            static_cast<uint64_t>(
-                std::ceil(double(buff_params.bytes) * fc_headroom_ratio)),
-            static_cast<uint32_t>(
-                std::ceil(double(buff_params.packets) * fc_headroom_ratio))};
-
         // Reconfigure flow control using the new frequency and headroom
         return _mgmt_portal->config_remote_stream(*_ctrl_xport,
             dst_epid,
             src_epid,
             lossy_xport,
-            fc_freq,
-            fc_headroom,
+            _get_buff_params_ratio(buff_params, fc_freq_ratio),
+            _get_buff_params_ratio(buff_params, fc_headroom_ratio),
             reset,
             STREAM_SETUP_TIMEOUT);
     }
 
     virtual chdr_tx_data_xport::uptr create_host_to_device_data_stream(
         const sep_addr_t dst_addr,
-        const bool lossy_xport,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const double fc_freq_ratio,
-        const double fc_headroom_ratio,
         const device_addr_t& xport_args)
     {
-        // Create a new source endpoint and EPID
-        sep_addr_t sw_epid_addr(_my_device_id, SEP_INST_DATA_BASE + (_data_ep_inst++));
-        sep_id_t src_epid = _epid_alloc->allocate_epid(sw_epid_addr);
-        _allocated_epids.insert(src_epid);
         _ensure_ep_is_reachable(dst_addr);
 
-        // Generate a new destination EPID instance
+        // Generate a new destination (device) EPID instance
         sep_id_t dst_epid = _epid_alloc->allocate_epid(dst_addr);
+        _mgmt_portal->initialize_endpoint(*_ctrl_xport, dst_addr, dst_epid);
 
-        // Create the data transport that we will return to the client
-        chdr_rx_data_xport::uptr xport = _mb_iface.make_rx_data_transport(
-            _my_device_id, src_epid, dst_epid, xport_args);
-
-        chdr_ctrl_xport::sptr mgmt_xport =
-            _mb_iface.make_ctrl_transport(_my_device_id, src_epid);
-
-        // Create new temporary management portal with the transports used for this stream
-        // TODO: This is a bit excessive. Maybe we can pair down the functionality of the
-        // portal just for route setup purposes. Whatever we do, we *must* use xport in it
-        // though otherwise the transport will not behave correctly.
-        mgmt_portal::uptr data_mgmt_portal =
-            mgmt_portal::make(*mgmt_xport, _pkt_factory, sw_epid_addr, src_epid);
-
-        // Setup a route to the EPID
-        data_mgmt_portal->initialize_endpoint(*mgmt_xport, dst_addr, dst_epid);
-        data_mgmt_portal->setup_local_route(*mgmt_xport, dst_epid);
         if (!_mgmt_portal->get_endpoint_info(dst_epid).has_data) {
             throw uhd::rfnoc_error("Downstream endpoint does not support data traffic");
         }
 
-        // TODO: Implement data transport setup logic here
-
+        // Create a new destination (host) endpoint and EPID
+        sep_addr_t sw_epid_addr(_my_device_id, SEP_INST_DATA_BASE + (_data_ep_inst++));
+        sep_id_t src_epid = _epid_alloc->allocate_epid(sw_epid_addr);
+        _allocated_epids.insert(src_epid);
 
-        // Delete the portal when done
-        data_mgmt_portal.reset();
-        return xport;
+        return _mb_iface.make_tx_data_transport({sw_epid_addr, dst_addr},
+            {src_epid, dst_epid},
+            pyld_buff_fmt,
+            mdata_buff_fmt,
+            xport_args);
     }
 
-    virtual chdr_tx_data_xport::uptr create_device_to_host_data_stream(
-        const sep_addr_t src_addr,
-        const bool lossy_xport,
+    virtual chdr_rx_data_xport::uptr create_device_to_host_data_stream(
+        sep_addr_t src_addr,
         const sw_buff_t pyld_buff_fmt,
         const sw_buff_t mdata_buff_fmt,
-        const double fc_freq_ratio,
-        const double fc_headroom_ratio,
         const device_addr_t& xport_args)
     {
-        // TODO: Implement me
-        return chdr_tx_data_xport::uptr();
+        _ensure_ep_is_reachable(src_addr);
+
+        // Generate a new source (device) EPID instance
+        sep_id_t src_epid = _epid_alloc->allocate_epid(src_addr);
+        _mgmt_portal->initialize_endpoint(*_ctrl_xport, src_addr, src_epid);
+
+        if (!_mgmt_portal->get_endpoint_info(src_epid).has_data) {
+            throw uhd::rfnoc_error("Downstream endpoint does not support data traffic");
+        }
+
+        // Create a new destination (host) endpoint and EPID
+        sep_addr_t sw_epid_addr(_my_device_id, SEP_INST_DATA_BASE + (_data_ep_inst++));
+        sep_id_t dst_epid = _epid_alloc->allocate_epid(sw_epid_addr);
+        _allocated_epids.insert(dst_epid);
+
+        return _mb_iface.make_rx_data_transport({src_addr, sw_epid_addr},
+            {src_epid, dst_epid},
+            pyld_buff_fmt,
+            mdata_buff_fmt,
+            xport_args);
     }
 
 private:
@@ -295,7 +280,14 @@ private:
         throw uhd::routing_error("Specified endpoint is not reachable");
     }
 
-    // A reference to the packet factor
+    stream_buff_params_t _get_buff_params_ratio(
+        const stream_buff_params_t& buff_params, const double ratio)
+    {
+        return {static_cast<uint64_t>(std::ceil(double(buff_params.bytes) * ratio)),
+            static_cast<uint32_t>(std::ceil(double(buff_params.packets) * ratio))};
+    }
+
+    // A reference to the packet factory
     const chdr::chdr_packet_factory& _pkt_factory;
     // The device address of this software endpoint
     const device_id_t _my_device_id;
diff --git a/host/lib/rfnoc/mgmt_portal.cpp b/host/lib/rfnoc/mgmt_portal.cpp
index b490e0baf..a8b72cbdf 100644
--- a/host/lib/rfnoc/mgmt_portal.cpp
+++ b/host/lib/rfnoc/mgmt_portal.cpp
@@ -794,12 +794,11 @@ private: // Functions
         mgmt_hop_t& hop)
     {
         // Validate flow control parameters
-        if (fc_freq.bytes >= (uint64_t(1) << 40)
-            || fc_freq.packets >= (uint64_t(1) << 24)) {
+        if (fc_freq.bytes > MAX_FC_FREQ_BYTES || fc_freq.packets > MAX_FC_FREQ_PKTS) {
             throw uhd::value_error("Flow control frequency parameters out of bounds");
         }
-        if (fc_headroom.bytes >= (uint64_t(1) << 16)
-            || fc_headroom.packets >= (uint64_t(1) << 8)) {
+        if (fc_headroom.bytes > MAX_FC_HEADROOM_BYTES
+            || fc_headroom.packets > MAX_FC_HEADROOM_PKTS) {
             throw uhd::value_error("Flow control headroom parameters out of bounds");
         }
 
@@ -992,7 +991,8 @@ private: // Functions
 
         auto send_buff = xport.get_send_buff(timeout * 1000);
         if (not send_buff) {
-            UHD_LOG_ERROR("RFNOC::MGMT", "Timed out getting send buff for management transaction");
+            UHD_LOG_ERROR(
+                "RFNOC::MGMT", "Timed out getting send buff for management transaction");
             throw uhd::io_error("Timed out getting send buff for management transaction");
         }
         _send_pkt->refresh(send_buff->data(), header, payload);
diff --git a/host/lib/rfnoc/rfnoc_graph.cpp b/host/lib/rfnoc/rfnoc_graph.cpp
index dd3dd7b90..4bf35cff1 100644
--- a/host/lib/rfnoc/rfnoc_graph.cpp
+++ b/host/lib/rfnoc/rfnoc_graph.cpp
@@ -15,12 +15,18 @@
 #include <uhdlib/rfnoc/graph.hpp>
 #include <uhdlib/rfnoc/graph_stream_manager.hpp>
 #include <uhdlib/rfnoc/rfnoc_device.hpp>
+#include <uhdlib/rfnoc/rfnoc_rx_streamer.hpp>
+#include <uhdlib/rfnoc/rfnoc_tx_streamer.hpp>
 #include <uhdlib/utils/narrow.hpp>
+#include <boost/make_shared.hpp>
 #include <boost/shared_ptr.hpp> // FIXME remove when rfnoc_device is ready
 #include <memory>
 
 using namespace uhd::rfnoc;
 
+namespace {
+const std::string LOG_ID("RFNOC::GRAPH");
+}
 
 class rfnoc_graph_impl : public rfnoc_graph
 {
@@ -33,6 +39,7 @@ public:
         , _graph(std::make_unique<uhd::rfnoc::detail::graph_t>())
     {
         setup_graph(dev_addr);
+        _init_sep_map();
         _init_static_connections();
     }
 
@@ -76,31 +83,136 @@ public:
         }
         if (!has_block(dst_blk)) {
             throw uhd::lookup_error(
-                std::string("Cannot connect blocks, source block not found: ")
-                + src_blk.to_string());
+                std::string("Cannot connect blocks, destination block not found: ")
+                + dst_blk.to_string());
         }
+        auto edge_type = _physical_connect(src_blk, src_port, dst_blk, dst_port);
         _connect(get_block(src_blk),
             src_port,
             get_block(dst_blk),
             dst_port,
+            edge_type,
             skip_property_propagation);
-        _physical_connect(src_blk, src_port, dst_blk, dst_port);
     }
 
-    void connect(uhd::tx_streamer& /*streamer*/,
-        size_t /*strm_port*/,
-        const block_id_t& /*dst_blk*/,
-        size_t /*dst_port*/)
+    void connect(uhd::tx_streamer::sptr streamer,
+        size_t strm_port,
+        const block_id_t& dst_blk,
+        size_t dst_port)
     {
-        throw uhd::not_implemented_error("");
+        // Verify the streamer was created by us
+        auto rfnoc_streamer = boost::dynamic_pointer_cast<rfnoc_tx_streamer>(streamer);
+        if (!rfnoc_streamer) {
+            throw uhd::type_error("Streamer is not rfnoc capable");
+        }
+
+        // Verify src_blk even exists in this graph
+        if (!has_block(dst_blk)) {
+            throw uhd::lookup_error(
+                std::string("Cannot connect block to streamer, source block not found: ")
+                + dst_blk.to_string());
+        }
+
+        // Verify src_blk has an SEP upstream
+        graph_edge_t dst_static_edge = _assert_edge(
+            _get_static_edge(
+                [dst_blk_id = dst_blk.to_string(), dst_port](const graph_edge_t& edge) {
+                    return edge.dst_blockid == dst_blk_id && edge.dst_port == dst_port;
+                }),
+            dst_blk.to_string());
+        if (block_id_t(dst_static_edge.src_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                dst_blk.to_string() + ":" + std::to_string(dst_port)
+                + " is not connected to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+
+        // Now get the name and address of the SEP
+        const std::string sep_block_id = dst_static_edge.src_blockid;
+        const sep_addr_t sep_addr      = _sep_map.at(sep_block_id);
+
+        const sw_buff_t pyld_fmt =
+            bits_to_sw_buff(rfnoc_streamer->get_otw_item_comp_bit_width());
+        const sw_buff_t mdata_fmt = BUFF_U64;
+
+        auto xport = _gsm->create_host_to_device_data_stream(sep_addr,
+            pyld_fmt,
+            mdata_fmt,
+            rfnoc_streamer->get_stream_args().args);
+
+        rfnoc_streamer->connect_channel(strm_port, std::move(xport));
+
+        //// If this worked, then also connect the streamer in the BGL graph
+        auto dst = get_block(dst_blk);
+        graph_edge_t edge_info(strm_port, dst_port, graph_edge_t::TX_STREAM, true);
+        _graph->connect(rfnoc_streamer.get(), dst.get(), edge_info);
     }
 
-    void connect(const block_id_t& /*src_blk*/,
-        size_t /*src_port*/,
-        uhd::rx_streamer& /*streamer*/,
-        size_t /*strm_port*/)
+    void connect(const block_id_t& src_blk,
+        size_t src_port,
+        uhd::rx_streamer::sptr streamer,
+        size_t strm_port)
     {
-        throw uhd::not_implemented_error("");
+        // Verify the streamer was created by us
+        auto rfnoc_streamer = boost::dynamic_pointer_cast<rfnoc_rx_streamer>(streamer);
+        if (!rfnoc_streamer) {
+            throw uhd::type_error("Streamer is not rfnoc capable");
+        }
+
+        // Verify src_blk even exists in this graph
+        if (!has_block(src_blk)) {
+            throw uhd::lookup_error(
+                std::string("Cannot connect block to streamer, source block not found: ")
+                + src_blk.to_string());
+        }
+
+        // Verify src_blk has an SEP downstream
+        graph_edge_t src_static_edge = _assert_edge(
+            _get_static_edge(
+                [src_blk_id = src_blk.to_string(), src_port](const graph_edge_t& edge) {
+                    return edge.src_blockid == src_blk_id && edge.src_port == src_port;
+                }),
+            src_blk.to_string());
+        if (block_id_t(src_static_edge.dst_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                src_blk.to_string() + ":" + std::to_string(src_port)
+                + " is not connected to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+
+        // Now get the name and address of the SEP
+        const std::string sep_block_id = src_static_edge.dst_blockid;
+        const sep_addr_t sep_addr      = _sep_map.at(sep_block_id);
+
+        const sw_buff_t pyld_fmt =
+            bits_to_sw_buff(rfnoc_streamer->get_otw_item_comp_bit_width());
+        const sw_buff_t mdata_fmt = BUFF_U64;
+
+        auto xport = _gsm->create_device_to_host_data_stream(sep_addr,
+            pyld_fmt,
+            mdata_fmt,
+            rfnoc_streamer->get_stream_args().args);
+
+        rfnoc_streamer->connect_channel(strm_port, std::move(xport));
+
+        // If this worked, then also connect the streamer in the BGL graph
+        auto src = get_block(src_blk);
+        graph_edge_t edge_info(src_port, strm_port, graph_edge_t::RX_STREAM, true);
+        _graph->connect(src.get(), rfnoc_streamer.get(), edge_info);
+    }
+
+    uhd::rx_streamer::sptr create_rx_streamer(
+        const size_t num_chans, const uhd::stream_args_t& args)
+    {
+        return boost::make_shared<rfnoc_rx_streamer>(num_chans, args);
+    }
+
+    uhd::tx_streamer::sptr create_tx_streamer(
+        const size_t num_chans, const uhd::stream_args_t& args)
+    {
+        return boost::make_shared<rfnoc_tx_streamer>(num_chans, args);
     }
 
     std::shared_ptr<mb_controller> get_mb_controller(const size_t mb_index = 0)
@@ -152,7 +264,7 @@ private:
             throw uhd::key_error(std::string("Found no RFNoC devices for ----->\n")
                                  + dev_addr.to_pp_string());
         }
-        _tree = _device->get_tree();
+        _tree        = _device->get_tree();
         _num_mboards = _tree->list("/mboards").size();
         for (size_t i = 0; i < _num_mboards; ++i) {
             _mb_controllers.emplace(i, _device->get_mb_controller(i));
@@ -170,7 +282,8 @@ private:
         try {
             _gsm = graph_stream_manager::make(pkt_factory, epid_alloc, links);
         } catch (uhd::io_error& ex) {
-            UHD_LOG_ERROR("RFNOC::GRAPH", "IO Error during GSM initialization. " << ex.what());
+            UHD_LOG_ERROR(
+                "RFNOC::GRAPH", "IO Error during GSM initialization. " << ex.what());
             throw;
         }
 
@@ -187,6 +300,9 @@ private:
             _gsm->connect_host_to_device(ctrl_sep_addr);
             // Grab and stash the Client Zero for this mboard
             detail::client_zero::sptr mb_cz = _gsm->get_client_zero(ctrl_sep_addr);
+            // Client zero port numbers are based on the control xbar numbers,
+            // which have the client 0 interface first, followed by stream
+            // endpoints, and then the blocks.
             _client_zeros.emplace(mb_idx, mb_cz);
 
             const size_t num_blocks       = mb_cz->get_num_blocks();
@@ -204,7 +320,7 @@ private:
 
             // Iterate through and register each of the blocks in this mboard
             for (size_t portno = 0; portno < num_blocks; ++portno) {
-                auto noc_id = mb_cz->get_noc_id(portno + first_block_port);
+                auto noc_id             = mb_cz->get_noc_id(portno + first_block_port);
                 auto block_factory_info = factory::get_block_factory(noc_id);
                 auto block_info = mb_cz->get_block_info(portno + first_block_port);
                 block_id_t block_id(mb_idx,
@@ -222,24 +338,25 @@ private:
                 //   iface object through the mb_iface
                 auto ctrlport_clk_iface =
                     mb.get_clock_iface(block_factory_info.ctrlport_clk);
-                auto tb_clk_iface = (block_factory_info.timebase_clk == CLOCK_KEY_GRAPH) ?
-                    std::make_shared<clock_iface>(CLOCK_KEY_GRAPH) :
-                    mb.get_clock_iface(block_factory_info.timebase_clk);
+                auto tb_clk_iface =
+                    (block_factory_info.timebase_clk == CLOCK_KEY_GRAPH)
+                        ? std::make_shared<clock_iface>(CLOCK_KEY_GRAPH)
+                        : mb.get_clock_iface(block_factory_info.timebase_clk);
                 // A "graph" clock is always "running"
                 if (block_factory_info.timebase_clk == CLOCK_KEY_GRAPH) {
                     tb_clk_iface->set_running(true);
                 }
-                auto block_reg_iface = _gsm->get_block_register_iface(ctrl_sep_addr,
+                auto block_reg_iface   = _gsm->get_block_register_iface(ctrl_sep_addr,
                     portno,
                     *ctrlport_clk_iface.get(),
                     *tb_clk_iface.get());
-                auto make_args_uptr = std::make_unique<noc_block_base::make_args_t>();
+                auto make_args_uptr    = std::make_unique<noc_block_base::make_args_t>();
                 make_args_uptr->noc_id = noc_id;
-                make_args_uptr->block_id = block_id;
-                make_args_uptr->num_input_ports = block_info.num_inputs;
-                make_args_uptr->num_output_ports = block_info.num_outputs;
-                make_args_uptr->reg_iface = block_reg_iface;
-                make_args_uptr->tb_clk_iface = tb_clk_iface;
+                make_args_uptr->block_id           = block_id;
+                make_args_uptr->num_input_ports    = block_info.num_inputs;
+                make_args_uptr->num_output_ports   = block_info.num_outputs;
+                make_args_uptr->reg_iface          = block_reg_iface;
+                make_args_uptr->tb_clk_iface       = tb_clk_iface;
                 make_args_uptr->ctrlport_clk_iface = ctrlport_clk_iface;
                 make_args_uptr->mb_control = (factory::has_requested_mb_access(noc_id)
                                                   ? _mb_controllers.at(mb_idx)
@@ -262,40 +379,43 @@ private:
         _block_registry->init_props();
     }
 
+    void _init_sep_map()
+    {
+        for (size_t mb_idx = 0; mb_idx < get_num_mboards(); ++mb_idx) {
+            auto remote_device_id = _device->get_mb_iface(mb_idx).get_remote_device_id();
+            auto& cz              = _client_zeros.at(mb_idx);
+            for (size_t sep_idx = 0; sep_idx < cz->get_num_stream_endpoints();
+                 ++sep_idx) {
+                // Register ID in _port_block_map
+                block_id_t id(mb_idx, NODE_ID_SEP, sep_idx);
+                _port_block_map.insert({{mb_idx, sep_idx + 1}, id});
+                _sep_map.insert({id.to_string(), sep_addr_t(remote_device_id, sep_idx)});
+            }
+        }
+    }
+
     void _init_static_connections()
     {
         UHD_LOG_TRACE("RFNOC::GRAPH", "Identifying static connections...");
         for (auto& kv_cz : _client_zeros) {
-            auto& adjacency_list          = kv_cz.second->get_adjacency_list();
-            const size_t first_block_port = 1 + kv_cz.second->get_num_stream_endpoints();
-
+            auto& adjacency_list = kv_cz.second->get_adjacency_list();
             for (auto& edge : adjacency_list) {
                 // Assemble edge
                 auto graph_edge = graph_edge_t();
-                if (edge.src_blk_index < first_block_port) {
-                    block_id_t id(kv_cz.first, NODE_ID_SEP, edge.src_blk_index - 1);
-                    _port_block_map.insert({{kv_cz.first, edge.src_blk_index}, id});
-                    graph_edge.src_blockid = id.to_string();
-                } else {
-                    graph_edge.src_blockid =
-                        _port_block_map.at({kv_cz.first, edge.src_blk_index});
-                }
-                if (edge.dst_blk_index < first_block_port) {
-                    block_id_t id(kv_cz.first, NODE_ID_SEP, edge.dst_blk_index - 1);
-                    _port_block_map.insert({{kv_cz.first, edge.dst_blk_index}, id});
-                    graph_edge.dst_blockid = id.to_string();
-                } else {
-                    graph_edge.dst_blockid =
-                        _port_block_map.at({kv_cz.first, edge.dst_blk_index});
-                }
+                UHD_ASSERT_THROW(
+                    _port_block_map.count({kv_cz.first, edge.src_blk_index}));
+                graph_edge.src_blockid =
+                    _port_block_map.at({kv_cz.first, edge.src_blk_index});
+                UHD_ASSERT_THROW(
+                    _port_block_map.count({kv_cz.first, edge.dst_blk_index}));
+                graph_edge.dst_blockid =
+                    _port_block_map.at({kv_cz.first, edge.dst_blk_index});
                 graph_edge.src_port = edge.src_blk_port;
                 graph_edge.dst_port = edge.dst_blk_port;
                 graph_edge.edge     = graph_edge_t::edge_t::STATIC;
                 _static_edges.push_back(graph_edge);
-                UHD_LOG_TRACE("RFNOC::GRAPH",
-                    "Static connection: "
-                        << graph_edge.src_blockid << ":" << graph_edge.src_port << " -> "
-                        << graph_edge.dst_blockid << ":" << graph_edge.dst_port);
+                UHD_LOG_TRACE(
+                    "RFNOC::GRAPH", "Static connection: " << graph_edge.to_string());
             }
         }
     }
@@ -312,214 +432,98 @@ private:
         size_t src_port,
         std::shared_ptr<node_t> dst_blk,
         size_t dst_port,
+        graph_edge_t::edge_t edge_type,
         bool skip_property_propagation)
     {
         graph_edge_t edge_info(
-            src_port, dst_port, graph_edge_t::DYNAMIC, not skip_property_propagation);
+            src_port, dst_port, edge_type, not skip_property_propagation);
         edge_info.src_blockid = src_blk->get_unique_id();
         edge_info.dst_blockid = dst_blk->get_unique_id();
         _graph->connect(src_blk.get(), dst_blk.get(), edge_info);
     }
 
-    /*! Helper method to find a stream endpoint connected to a block
-     *
-     * \param blk_id the block connected to the stream endpoint
-     * \param port the port connected to the stream endpoint
-     * \param blk_is_src true if the block is a data source, false if it is a
-     *                   destination
-     * \return the address of the stream endpoint, or boost::none if it is not
-     *         directly connected to a stream endpoint
-     */
-    boost::optional<sep_addr_t> _get_adjacent_sep(
-        const block_id_t& blk_id, const size_t port, const bool blk_is_src) const
-    {
-        const std::string block_id_str = get_block(blk_id)->get_block_id().to_string();
-        UHD_LOG_TRACE("RFNOC::GRAPH",
-            "Finding SEP for " << (blk_is_src ? "source" : "dst") << " block "
-                               << block_id_str << ":" << port);
-        // TODO: This is an attempt to simplify the algo, but it turns out to be
-        // as many lines as before:
-        //auto edge_predicate = [blk_is_src, block_id_str](const graph_edge_t edge) {
-            //if (blk_is_src) {
-                //return edge.src_blockid == block_id_str;
-            //}
-            //return edge.dst_blockid == block_id_str;
-        //};
-
-        //auto blk_edge_it =
-            //std::find_if(_static_edges.cbegin(), _static_edges.cend(), edge_predicate);
-        //if (blk_edge_it == _static_edges.cend()) {
-            //return boost::none;
-        //}
-
-        //const std::string sep_block_id = blk_is_src ?
-            //blk_edge_it->dst_blockid : blk_edge_it->src_blockid;
-        //UHD_LOG_TRACE("RFNOC::GRAPH",
-            //"Found SEP: " << sep_block_id);
-
-        //auto port_map_result = std::find_if(_port_block_map.cbegin(),
-            //_port_block_map.cend,
-            //[sep_block_id](std::pair<std::pair<size_t, size_t>, block_id_t> port_block) {
-                //return port_block.second == sep_block_id;
-            //});
-        //if (port_map_result == _port_block_map.cend()) {
-            //throw uhd::lookup_error(
-                //std::string("SEP `") + sep_block_id + "' not found in port/block map!");
-        //}
-        //const auto dev = _device->get_mb_iface(mb_idx).get_remote_device_id();
-        //const sep_inst_t sep_inst = blk_is_src ?
-            //edge.dst_blk_index - 1 : edge.src_blk_index - 1;
-        //return sep_addr_t(dev, sep_inst);
-
-        const auto& info  = _xbar_block_config.at(block_id_str);
-        const auto mb_idx = blk_id.get_device_no();
-        const auto cz     = _client_zeros.at(mb_idx);
-
-        const size_t first_block_port = 1 + cz->get_num_stream_endpoints();
-
-        for (const auto& edge : cz->get_adjacency_list()) {
-            const auto edge_blk_idx = blk_is_src ? edge.src_blk_index
-                                                 : edge.dst_blk_index;
-            const auto edge_blk_port = blk_is_src ? edge.src_blk_port : edge.dst_blk_port;
-
-            if ((edge_blk_idx == info.xbar_port + first_block_port)
-                and (edge_blk_port == port)) {
-                UHD_LOGGER_DEBUG("RFNOC::GRAPH")
-                    << boost::format("Block found in adjacency list. %d:%d->%d:%d")
-                           % edge.src_blk_index
-                           % static_cast<unsigned int>(edge.src_blk_port)
-                           % edge.dst_blk_index
-                           % static_cast<unsigned int>(edge.dst_blk_port);
-
-                // Check that the block is connected to a stream endpoint. The
-                // minus one here is because index zero is client 0.
-                const sep_inst_t sep_inst = blk_is_src ?
-                    edge.dst_blk_index - 1 : edge.src_blk_index - 1;
-
-                if (sep_inst < cz->get_num_stream_endpoints()) {
-                    const auto dev = _device->get_mb_iface(mb_idx).get_remote_device_id();
-                    return sep_addr_t(dev, sep_inst);
-                } else {
-                    // Block is connected to another block
-                    return boost::none;
-                }
-            }
-        }
-        return boost::none;
-    }
-
     /*! Internal physical connection helper
      *
      * Make the connections in the physical device
      *
-     * \throws connect_disallowed_on_src
-     *     if the source port is statically connected to a *different* block
-     * \throws connect_disallowed_on_dst
-     *     if the destination port is statically connected to a *different* block
+     * \throws uhd::routing_error
+     *     if the blocks are statically connected to something else
      */
-    void _physical_connect(const block_id_t& src_blk,
+    graph_edge_t::edge_t _physical_connect(const block_id_t& src_blk,
         size_t src_port,
         const block_id_t& dst_blk,
         size_t dst_port)
     {
-        auto src_blk_ctrl = get_block(src_blk);
-        auto dst_blk_ctrl = get_block(dst_blk);
-
-        /*
-         * Start by determining if the connection can be made
-         * Get the adjacency list and check if the connection is in it already
-         */
-        // Read the adjacency list for the source and destination blocks
-        auto src_mb_idx = src_blk.get_device_no();
-        auto src_cz     = _gsm->get_client_zero(
-            sep_addr_t(_device->get_mb_iface(src_mb_idx).get_remote_device_id(), 0));
-        std::vector<detail::client_zero::edge_def_t>& adj_list =
-            src_cz->get_adjacency_list();
-        // Check the src_blk
-        auto src_blk_xbar_info =
-            _xbar_block_config.at(src_blk_ctrl->get_block_id().to_string());
-        // This "xbar_port" starts at the first block, so we need to add the client zero
-        // and stream endpoint ports
-        const auto src_xbar_port =
-            src_blk_xbar_info.xbar_port + src_cz->get_num_stream_endpoints() + 1;
-        // We can also find out which stream endpoint the src block is connected to
-        sep_inst_t src_sep;
-        for (detail::client_zero::edge_def_t edge : adj_list) {
-            if ((edge.src_blk_index == src_xbar_port)
-                and (edge.src_blk_port == src_port)) {
-                UHD_LOGGER_DEBUG("RFNOC::GRAPH")
-                    << boost::format("Block found in adjacency list. %d:%d->%d:%d")
-                           % edge.src_blk_index
-                           % static_cast<unsigned int>(edge.src_blk_port)
-                           % edge.dst_blk_index
-                           % static_cast<unsigned int>(edge.dst_blk_port);
-                if (edge.dst_blk_index <= src_cz->get_num_stream_endpoints()) {
-                    src_sep =
-                        edge.dst_blk_index - 1 /* minus 1 because port 0 is client zero*/;
-                } else {
-                    // TODO connect_disallowed_on_src?
-                    // TODO put more info in exception
-                    throw uhd::routing_error(
-                        "Unable to connect to statically connected source port");
-                }
-            }
+        const std::string src_blk_info =
+            src_blk.to_string() + ":" + std::to_string(src_port);
+        const std::string dst_blk_info =
+            dst_blk.to_string() + ":" + std::to_string(dst_port);
+
+        // Find the static edge for src_blk:src_port
+        graph_edge_t src_static_edge = _assert_edge(
+            _get_static_edge(
+                [src_blk_id = src_blk.to_string(), src_port](const graph_edge_t& edge) {
+                    return edge.src_blockid == src_blk_id && edge.src_port == src_port;
+                }),
+            src_blk_info);
+
+        // Now see if it's already connected to the destination
+        if (src_static_edge.dst_blockid == dst_blk.to_string()
+            && src_static_edge.dst_port == dst_port) {
+            UHD_LOG_TRACE(LOG_ID,
+                "Blocks " << src_blk_info << " and " << dst_blk_info
+                          << " are already statically connected, no physical connection "
+                             "required.");
+            return graph_edge_t::STATIC;
         }
 
-        // Read the dst adjacency list if its different
-        // TODO they may be on the same mboard, which would make this redundant
-        auto dst_mb_idx = dst_blk.get_device_no();
-        auto dst_cz     = _gsm->get_client_zero(
-            sep_addr_t(_device->get_mb_iface(dst_mb_idx).get_remote_device_id(), 0));
-        adj_list = dst_cz->get_adjacency_list();
-        // Check the dst blk
-        auto dst_blk_xbar_info =
-            _xbar_block_config.at(dst_blk_ctrl->get_block_id().to_string());
-        // This "xbar_port" starts at the first block, so we need to add the client zero
-        // and stream endpoint ports
-        const auto dst_xbar_port =
-            dst_blk_xbar_info.xbar_port + dst_cz->get_num_stream_endpoints() + 1;
-        // We can also find out which stream endpoint the dst block is connected to
-        sep_inst_t dst_sep;
-        for (detail::client_zero::edge_def_t edge : adj_list) {
-            if ((edge.dst_blk_index == dst_xbar_port)
-                and (edge.dst_blk_port == dst_port)) {
-                UHD_LOGGER_DEBUG("RFNOC::GRAPH")
-                    << boost::format("Block found in adjacency list. %d:%d->%d:%d")
-                           % edge.src_blk_index
-                           % static_cast<unsigned int>(edge.src_blk_port)
-                           % edge.dst_blk_index
-                           % static_cast<unsigned int>(edge.dst_blk_port);
-                if (edge.src_blk_index <= dst_cz->get_num_stream_endpoints()) {
-                    dst_sep =
-                        edge.src_blk_index - 1 /* minus 1 because port 0 is client zero*/;
-                } else {
-                    // TODO connect_disallowed_on_dst?
-                    // TODO put more info in exception
-                    throw uhd::routing_error(
-                        "Unable to connect to statically connected destination port");
-                }
-            }
+        // If they're not statically connected, the source *must* be connected
+        // to an SEP, or this route is impossible
+        if (block_id_t(src_static_edge.dst_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                src_blk_info + " is neither statically connected to " + dst_blk_info
+                + " nor to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
         }
 
-        /* TODO: we checked if either port is used in a static connection (and its not if
-         * we've made it this far). We also need to check something else, but I can't
-         * remember what...
-         */
-
-        // At this point, we know the attempted connection is valid, so let's go ahead and
-        // make it
-        sep_addr_t src_sep_addr(
-            _device->get_mb_iface(src_mb_idx).get_remote_device_id(), src_sep);
-        sep_addr_t dst_sep_addr(
-            _device->get_mb_iface(dst_mb_idx).get_remote_device_id(), dst_sep);
+        // OK, now we know which source SEP we have
+        const std::string src_sep_info = src_static_edge.dst_blockid;
+        const sep_addr_t src_sep_addr  = _sep_map.at(src_sep_info);
+
+        // Now find the static edge for the destination SEP
+        auto dst_static_edge = _assert_edge(
+            _get_static_edge(
+                [dst_blk_id = dst_blk.to_string(), dst_port](const graph_edge_t& edge) {
+                    return edge.dst_blockid == dst_blk_id && edge.dst_port == dst_port;
+                }),
+            dst_blk_info);
+
+        // If they're not statically connected, the source *must* be connected
+        // to an SEP, or this route is impossible
+        if (block_id_t(dst_static_edge.src_blockid).get_block_name() != NODE_ID_SEP) {
+            const std::string err_msg =
+                dst_blk_info + " is neither statically connected to " + src_blk_info
+                + " nor to an SEP! Routing impossible.";
+            UHD_LOG_ERROR(LOG_ID, err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+
+        // OK, now we know which destination SEP we have
+        const std::string dst_sep_info = dst_static_edge.src_blockid;
+        const sep_addr_t dst_sep_addr  = _sep_map.at(dst_sep_info);
+
+        // Now all we need to do is dynamically connect those SEPs
         auto strm_info = _gsm->create_device_to_device_data_stream(
             dst_sep_addr, src_sep_addr, false, 0.1, 0.0, false);
 
-        UHD_LOGGER_INFO("RFNOC::GRAPH")
+        UHD_LOGGER_DEBUG(LOG_ID)
             << boost::format("Data stream between EPID %d and EPID %d established "
                              "where downstream buffer can hold %lu bytes and %u packets")
                    % std::get<0>(strm_info).first % std::get<0>(strm_info).second
                    % std::get<1>(strm_info).bytes % std::get<1>(strm_info).packets;
+
+        return graph_edge_t::DYNAMIC;
     }
 
     //! Flush and reset each connected port on the mboard
@@ -541,6 +545,35 @@ private:
             mb_cz->reset_ctrl(block_portno);
         }
     }
+
+    /*! Find the static edge that matches \p pred
+     *
+     * \throws uhd::assertion_error if the edge can't be found. So be careful!
+     */
+    template <typename UnaryPredicate>
+    boost::optional<graph_edge_t> _get_static_edge(UnaryPredicate&& pred)
+    {
+        auto edge_it = std::find_if(_static_edges.cbegin(), _static_edges.cend(), pred);
+        if (edge_it == _static_edges.cend()) {
+            return boost::none;
+        }
+        return *edge_it;
+    }
+
+    /*! Make sure an optional edge info is valid, or throw.
+     */
+    graph_edge_t _assert_edge(
+        boost::optional<graph_edge_t> edge_o, const std::string& blk_info)
+    {
+        if (!bool(edge_o)) {
+            const std::string err_msg = std::string("Cannot connect block ") + blk_info
+                                        + ", port is unconnected in the FPGA!";
+            UHD_LOG_ERROR("RFNOC::GRAPH", err_msg);
+            throw uhd::routing_error(err_msg);
+        }
+        return edge_o.get();
+    }
+
     /**************************************************************************
      * Attributes
      *************************************************************************/
@@ -580,6 +613,9 @@ private:
     // or SEP
     std::map<std::pair<size_t, size_t>, block_id_t> _port_block_map;
 
+    //! Map SEP block ID (e.g. 0/SEP#0) onto a sep_addr_t
+    std::unordered_map<std::string, sep_addr_t> _sep_map;
+
     //! List of statically connected edges. Includes SEPs too!
     std::vector<graph_edge_t> _static_edges;
 
diff --git a/host/lib/rfnoc/rfnoc_rx_streamer.cpp b/host/lib/rfnoc/rfnoc_rx_streamer.cpp
new file mode 100644
index 000000000..4340faff0
--- /dev/null
+++ b/host/lib/rfnoc/rfnoc_rx_streamer.cpp
@@ -0,0 +1,141 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#include <uhd/rfnoc/defaults.hpp>
+#include <uhdlib/rfnoc/node_accessor.hpp>
+#include <uhdlib/rfnoc/rfnoc_rx_streamer.hpp>
+#include <atomic>
+
+using namespace uhd;
+using namespace uhd::rfnoc;
+
+const std::string STREAMER_ID = "RxStreamer";
+static std::atomic<uint64_t> streamer_inst_ctr;
+
+rfnoc_rx_streamer::rfnoc_rx_streamer(const size_t num_chans,
+    const uhd::stream_args_t stream_args)
+    : rx_streamer_impl<chdr_rx_data_xport>(num_chans, stream_args)
+    , _unique_id(STREAMER_ID + "#" + std::to_string(streamer_inst_ctr++))
+    , _stream_args(stream_args)
+{
+    // No block to which to forward properties or actions
+    set_prop_forwarding_policy(forwarding_policy_t::DROP);
+    set_action_forwarding_policy(forwarding_policy_t::DROP);
+
+    // Initialize properties
+    _scaling_in.reserve(num_chans);
+    _samp_rate_in.reserve(num_chans);
+    _tick_rate_in.reserve(num_chans);
+    _type_in.reserve(num_chans);
+
+    for (size_t i = 0; i < num_chans; i++) {
+        _register_props(i, stream_args.otw_format);
+    }
+    node_accessor_t node_accessor{};
+    node_accessor.init_props(this);
+}
+
+std::string rfnoc_rx_streamer::get_unique_id() const
+{
+    return _unique_id;
+}
+
+size_t rfnoc_rx_streamer::get_num_input_ports() const
+{
+    return get_num_channels();
+}
+
+size_t rfnoc_rx_streamer::get_num_output_ports() const
+{
+    return 0;
+}
+
+void rfnoc_rx_streamer::issue_stream_cmd(const stream_cmd_t& stream_cmd)
+{
+    if (get_num_channels() > 1 and stream_cmd.stream_now
+        and stream_cmd.stream_mode != stream_cmd_t::STREAM_MODE_STOP_CONTINUOUS) {
+        throw uhd::runtime_error(
+            "Invalid recv stream command - stream now on multiple channels in a "
+            "single streamer will fail to time align.");
+    }
+
+    auto cmd = stream_cmd_action_info::make(stream_cmd.stream_mode);
+    cmd->stream_cmd = stream_cmd;
+
+    for (size_t i = 0; i < get_num_channels(); i++) {
+        const res_source_info info(res_source_info::INPUT_EDGE, i);
+        post_action(info, cmd);
+    }
+}
+
+const uhd::stream_args_t& rfnoc_rx_streamer::get_stream_args() const
+{
+    return _stream_args;
+}
+
+bool rfnoc_rx_streamer::check_topology(
+    const std::vector<size_t>& connected_inputs,
+    const std::vector<size_t>& connected_outputs)
+{
+    // Check that all channels are connected
+    if (connected_inputs.size() != get_num_input_ports()) {
+        return false;
+    }
+
+    // Call base class to check that connections are valid
+    return node_t::check_topology(connected_inputs, connected_outputs);
+}
+
+void rfnoc_rx_streamer::_register_props(const size_t chan,
+    const std::string& otw_format)
+{
+    // Create actual properties and store them
+    _scaling_in.push_back(property_t<double>(
+        PROP_KEY_SCALING, {res_source_info::INPUT_EDGE, chan}));
+    _samp_rate_in.push_back(
+        property_t<double>(PROP_KEY_SAMP_RATE, {res_source_info::INPUT_EDGE, chan}));
+    _tick_rate_in.push_back(property_t<double>(
+        PROP_KEY_TICK_RATE, {res_source_info::INPUT_EDGE, chan}));
+    _type_in.emplace_back(property_t<std::string>(
+        PROP_KEY_TYPE, otw_format, {res_source_info::INPUT_EDGE, chan}));
+
+    // Give us some shorthands for the rest of this function
+    property_t<double>* scaling_in   = &_scaling_in.back();
+    property_t<double>* samp_rate_in = &_samp_rate_in.back();
+    property_t<double>* tick_rate_in = &_tick_rate_in.back();
+    property_t<std::string>* type_in = &_type_in.back();
+
+    // Register them
+    register_property(scaling_in);
+    register_property(samp_rate_in);
+    register_property(tick_rate_in);
+    register_property(type_in);
+
+    // Add resolvers
+    add_property_resolver({scaling_in}, {},
+        [&scaling_in = *scaling_in, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `scaling_in'@" << chan);
+            if (scaling_in.is_valid()) {
+                this->set_scale_factor(chan, scaling_in.get() / 32767.0);
+            }
+        });
+
+    add_property_resolver({samp_rate_in}, {},
+        [&samp_rate_in = *samp_rate_in, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `samp_rate_in'@" << chan);
+            if (samp_rate_in.is_valid()) {
+                this->set_samp_rate(samp_rate_in.get());
+            }
+        });
+
+    add_property_resolver({tick_rate_in}, {},
+        [&tick_rate_in = *tick_rate_in, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `tick_rate_in'@" << chan);
+            if (tick_rate_in.is_valid()) {
+                this->set_tick_rate(tick_rate_in.get());
+            }
+        });
+}
diff --git a/host/lib/rfnoc/rfnoc_tx_streamer.cpp b/host/lib/rfnoc/rfnoc_tx_streamer.cpp
new file mode 100644
index 000000000..82feeaf1f
--- /dev/null
+++ b/host/lib/rfnoc/rfnoc_tx_streamer.cpp
@@ -0,0 +1,124 @@
+//
+// Copyright 2019 Ettus Research, a National Instruments Brand
+//
+// SPDX-License-Identifier: GPL-3.0-or-later
+//
+
+#include <uhd/rfnoc/defaults.hpp>
+#include <uhdlib/rfnoc/node_accessor.hpp>
+#include <uhdlib/rfnoc/rfnoc_tx_streamer.hpp>
+#include <atomic>
+
+using namespace uhd;
+using namespace uhd::rfnoc;
+
+const std::string STREAMER_ID = "TxStreamer";
+static std::atomic<uint64_t> streamer_inst_ctr;
+
+rfnoc_tx_streamer::rfnoc_tx_streamer(const size_t num_chans,
+    const uhd::stream_args_t stream_args)
+    : tx_streamer_impl<chdr_tx_data_xport>(num_chans, stream_args)
+    , _unique_id(STREAMER_ID + "#" + std::to_string(streamer_inst_ctr++))
+    , _stream_args(stream_args)
+{
+    // No block to which to forward properties or actions
+    set_prop_forwarding_policy(forwarding_policy_t::DROP);
+    set_action_forwarding_policy(forwarding_policy_t::DROP);
+
+    // Initialize properties
+    _scaling_out.reserve(num_chans);
+    _samp_rate_out.reserve(num_chans);
+    _tick_rate_out.reserve(num_chans);
+    _type_out.reserve(num_chans);
+
+    for (size_t i = 0; i < num_chans; i++) {
+        _register_props(i, stream_args.otw_format);
+    }
+
+    node_accessor_t node_accessor;
+    node_accessor.init_props(this);
+}
+
+std::string rfnoc_tx_streamer::get_unique_id() const
+{
+    return _unique_id;
+}
+
+size_t rfnoc_tx_streamer::get_num_input_ports() const
+{
+    return 0;
+}
+
+size_t rfnoc_tx_streamer::get_num_output_ports() const
+{
+    return get_num_channels();
+}
+
+const uhd::stream_args_t& rfnoc_tx_streamer::get_stream_args() const
+{
+    return _stream_args;
+}
+
+bool rfnoc_tx_streamer::check_topology(
+    const std::vector<size_t>& connected_inputs,
+    const std::vector<size_t>& connected_outputs)
+{
+    // Check that all channels are connected
+    if (connected_outputs.size() != get_num_output_ports()) {
+        return false;
+    }
+
+    // Call base class to check that connections are valid
+    return node_t::check_topology(connected_inputs, connected_outputs);
+}
+
+void rfnoc_tx_streamer::_register_props(const size_t chan,
+    const std::string& otw_format)
+{
+    // Create actual properties and store them
+    _scaling_out.push_back(property_t<double>(
+        PROP_KEY_SCALING, {res_source_info::OUTPUT_EDGE, chan}));
+    _samp_rate_out.push_back(property_t<double>(
+        PROP_KEY_SAMP_RATE, {res_source_info::OUTPUT_EDGE, chan}));
+    _tick_rate_out.push_back(property_t<double>(
+        PROP_KEY_TICK_RATE, {res_source_info::OUTPUT_EDGE, chan}));
+    _type_out.emplace_back(property_t<std::string>(
+        PROP_KEY_TYPE, otw_format, {res_source_info::OUTPUT_EDGE, chan}));
+
+    // Give us some shorthands for the rest of this function
+    property_t<double>* scaling_out   = &_scaling_out.back();
+    property_t<double>* samp_rate_out = &_samp_rate_out.back();
+    property_t<double>* tick_rate_out = &_tick_rate_out.back();
+    property_t<std::string>* type_out = &_type_out.back();
+
+    // Register them
+    register_property(scaling_out);
+    register_property(samp_rate_out);
+    register_property(tick_rate_out);
+    register_property(type_out);
+
+    // Add resolvers
+    add_property_resolver({scaling_out}, {},
+        [&scaling_out = *scaling_out, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `scaling_out'@" << chan);
+            if (scaling_out.is_valid()) {
+                this->set_scale_factor(chan, 32767.0 / scaling_out.get());
+            }
+        });
+
+    add_property_resolver({samp_rate_out}, {},
+        [&samp_rate_out = *samp_rate_out, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `samp_rate_out'@" << chan);
+            if (samp_rate_out.is_valid()) {
+                this->set_samp_rate(samp_rate_out.get());
+            }
+        });
+
+    add_property_resolver({tick_rate_out}, {},
+        [&tick_rate_out = *tick_rate_out, chan, this]() {
+            RFNOC_LOG_TRACE("Calling resolver for `tick_rate_out'@" << chan);
+            if (tick_rate_out.is_valid()) {
+                this->set_tick_rate(tick_rate_out.get());
+            }
+        });
+}