RFNoC: Fix graph traversal

- Allows different TX sample rates on separate channels
- Prevents stream commands from being issued on wrong ports
- Prevents some receive timeout errors

4 files changed, 56 insertions, 24 deletions

diff --git a/host/include/uhd/rfnoc/node_ctrl_base.ipp b/host/include/uhd/rfnoc/node_ctrl_base.ipp
index 79a8fe2a9..266df445a 100644
--- a/host/include/uhd/rfnoc/node_ctrl_base.ipp
+++ b/host/include/uhd/rfnoc/node_ctrl_base.ipp
@@ -27,27 +27,42 @@ namespace uhd {
         // List of return values:
         std::set< T_sptr > results_s;
         // To avoid cycles:
-        std::set< sptr > explored;
+        std::set < sptr > explored;
         // Initialize our search queue with ourself:
-        std::set< sptr > search_q;
-        search_q.insert(shared_from_this());
-        std::set< sptr > next_q;
+        std::set < node_map_pair_t > search_q;
+        // FIXME:  Port is initialized to ANY_PORT, but it should really be
+        // passed in by the caller.
+        search_q.insert(node_map_pair_t(ANY_PORT, shared_from_this()));
+        std::set < node_map_pair_t > next_q;
 
         while (iters++ < MAX_ITER) {
             next_q.clear();
-            BOOST_FOREACH(const sptr &this_node, search_q) {
+            BOOST_FOREACH(const node_map_pair_t node_pair, search_q) {
+                sptr node = node_pair.second.lock();
+                if (not node)
+                {
+                    continue;
+                }
+                size_t our_port = node_pair.first;
                 // Add this node to the list of explored nodes
-                explored.insert(this_node);
+                explored.insert(node);
                 // Create set of all child nodes of this_node that are not in explored:
-                std::set< sptr > next_nodes;
+                std::set< node_map_pair_t > next_nodes;
                 {
-                    node_map_t all_next_nodes = downstream ? this_node->list_downstream_nodes() : this_node->list_upstream_nodes();
+                    node_map_t all_next_nodes = downstream ?
+                        node->list_downstream_nodes() :
+                        node->list_upstream_nodes();
                     for (
                         node_map_t::iterator it = all_next_nodes.begin();
                         it != all_next_nodes.end();
                         ++it
                     ) {
-                        size_t our_port = it->first;
+                        size_t connected_port = it->first;
+                        // If port is given, limit traversal to only that port.
+                        if (our_port != ANY_PORT and our_port != connected_port)
+                        {
+                            continue;
+                        }
                         if (active_only
                             and not (downstream ? _tx_streamer_active[our_port] : _rx_streamer_active[our_port] )) {
                             continue;
@@ -60,7 +75,22 @@ namespace uhd {
                         if (next_node_sptr) {
                             results_s.insert(next_node_sptr);
                         } else {
-                            next_nodes.insert(one_next_node);
+                            size_t next_port = ANY_PORT;
+                            // FIXME:  Need proper mapping from input port
+                            // to output port.
+                            // The code below assumes that blocks with the same
+                            // number of connected upstream and downstream nodes
+                            // map each input port directly to the same output
+                            // port.  This limits the graph traversal to prevent
+                            // finding nodes that are not part of this chain.
+                            if (one_next_node->list_upstream_nodes().size() ==
+                                one_next_node->list_downstream_nodes().size())
+                            {
+                                next_port = (downstream ?
+                                    node->get_downstream_port(connected_port) :
+                                    node->get_upstream_port(connected_port));
+                            }
+                            next_nodes.insert(node_map_pair_t(next_port, it->second));
                         }
                     }
                 }
diff --git a/host/lib/rfnoc/ddc_block_ctrl_impl.cpp b/host/lib/rfnoc/ddc_block_ctrl_impl.cpp
index b70ae8959..7ea81e255 100644
--- a/host/lib/rfnoc/ddc_block_ctrl_impl.cpp
+++ b/host/lib/rfnoc/ddc_block_ctrl_impl.cpp
@@ -186,14 +186,7 @@ public:
             stream_cmd.num_samps *= decimation;
         }
 
-        source_node_ctrl::sptr this_upstream_block_ctrl =
-                boost::dynamic_pointer_cast<source_node_ctrl>(list_upstream_nodes().at(chan).lock());
-        if (this_upstream_block_ctrl) {
-            this_upstream_block_ctrl->issue_stream_cmd(
-                    stream_cmd,
-                    get_upstream_port(chan)
-            );
-        }
+        source_block_ctrl_base::issue_stream_cmd(stream_cmd, chan);
     }
 
 private:
diff --git a/host/lib/rfnoc/duc_block_ctrl_impl.cpp b/host/lib/rfnoc/duc_block_ctrl_impl.cpp
index cc9d30361..cf22ab7da 100644
--- a/host/lib/rfnoc/duc_block_ctrl_impl.cpp
+++ b/host/lib/rfnoc/duc_block_ctrl_impl.cpp
@@ -173,11 +173,7 @@ public:
             stream_cmd.num_samps *= interpolation;
         }
 
-        for(const node_ctrl_base::node_map_pair_t upstream_node:  list_upstream_nodes()) {
-            source_node_ctrl::sptr this_upstream_block_ctrl =
-                boost::dynamic_pointer_cast<source_node_ctrl>(upstream_node.second.lock());
-            this_upstream_block_ctrl->issue_stream_cmd(stream_cmd, chan);
-        }
+        source_block_ctrl_base::issue_stream_cmd(stream_cmd, chan);
     }
 
 private:
diff --git a/host/lib/rfnoc/source_block_ctrl_base.cpp b/host/lib/rfnoc/source_block_ctrl_base.cpp
index 0f1c31e4f..364aa01c6 100644
--- a/host/lib/rfnoc/source_block_ctrl_base.cpp
+++ b/host/lib/rfnoc/source_block_ctrl_base.cpp
@@ -29,9 +29,22 @@ void source_block_ctrl_base::issue_stream_cmd(
     }
 
     for(const node_ctrl_base::node_map_pair_t upstream_node:  _upstream_nodes) {
+        // FIXME:  Need proper mapping from input port to output port
+        // The code below assumes the input port and output port are the same
+        // if the number of upstream and downstream connections are the same.
+        // The stream command is limited to only that port to prevent issuing
+        // it on the wrong block and port.
+        if (_upstream_nodes.size() == _downstream_nodes.size() and
+            upstream_node.first != chan)
+        {
+            continue;
+        }
         source_node_ctrl::sptr this_upstream_block_ctrl =
             boost::dynamic_pointer_cast<source_node_ctrl>(upstream_node.second.lock());
-        this_upstream_block_ctrl->issue_stream_cmd(stream_cmd, chan);
+        if (this_upstream_block_ctrl)
+        {
+            this_upstream_block_ctrl->issue_stream_cmd(stream_cmd, get_upstream_port(chan));
+        }
     }
 }