runtime: fix disconnecting a msg port on a hier_block2

It seems that when you call msg_connect, the call to message_port_sub is not made until the flowgraph has been flattened, so the subscription does not belong to the hier block, it belongs to the block inside the hier block that has the message output. However, when you call msg_disconnect, it simply attempts to call message_port_unsub on on the hier block, which fails, as the hier block has no subscriptions." The fix simply checks if the source or destination blocks are hierarchical, and if they are, resolves which block in the hierarchical block has the message port, then calls message_port_unsub with that block.
author: Mark Cottrell <mark.cottrell@taitradio.com> 2013-10-15 09:05:21 -0700
committer: Johnathan Corgan <johnathan@corganlabs.com> 2013-10-15 09:07:40 -0700
commit: cd19c5a64d14e4cbb90e08648aa45cd76543deed (patch)
tree: f638e92943556229cdf7f1fc450ed0b00bbcbf4b
parent: d5f0cec440ef2848992ce0ae34bfe73bb0c9772a (diff)
2 files changed, 235 insertions, 4 deletions
diff --git a/gnuradio-runtime/lib/hier_block2_detail.cc b/gnuradio-runtime/lib/hier_block2_detail.cc
index ebfaa6fa69..1f5f5b0738 100644
--- a/gnuradio-runtime/lib/hier_block2_detail.cc
+++ b/gnuradio-runtime/lib/hier_block2_detail.cc
@@ -197,14 +197,41 @@ namespace gr {
   {
     if(HIER_BLOCK2_DETAIL_DEBUG)
       std::cout << "disconnecting message port..." << std::endl;
-    
-    // unregister the subscription - if already subscribed
-    src->message_port_unsub(srcport, pmt::cons(dst->alias_pmt(), dstport));
 
     // remove edge for this message connection
-    bool hier_out = (d_owner == src.get()) && src->message_port_is_hier_out(srcport);;
+    bool hier_out = (d_owner == src.get()) && src->message_port_is_hier_out(srcport);
     bool hier_in = (d_owner == dst.get()) && dst->message_port_is_hier_in(dstport);
+
     d_fg->disconnect(msg_endpoint(src, srcport, hier_out), msg_endpoint(dst, dstport, hier_in));
+
+    hier_block2_sptr src_block(cast_to_hier_block2_sptr(src));
+    hier_block2_sptr dst_block(cast_to_hier_block2_sptr(dst));
+
+    if (src_block && src.get() != d_owner) {
+        // if the source is hier, we need to resolve the endpoint before calling unsub
+        msg_edge_vector_t edges = src_block->d_detail->d_fg->msg_edges();
+        for (msg_edge_viter_t it = edges.begin(); it != edges.end(); ++it) {
+            if ((*it).dst().block() == src) {
+                src = (*it).src().block();
+                srcport = (*it).src().port();
+            }
+        }
+    }
+
+    if (dst_block && dst.get() != d_owner) {
+        // if the destination is hier, we need to resolve the endpoint before calling unsub
+        msg_edge_vector_t edges = dst_block->d_detail->d_fg->msg_edges();
+        for (msg_edge_viter_t it = edges.begin(); it != edges.end(); ++it) {
+            if ((*it).src().block() == dst) {
+                dst = (*it).dst().block();
+                dstport = (*it).dst().port();
+            }
+        }
+    }
+
+    // unregister the subscription - if already subscribed
+    src->message_port_unsub(srcport, pmt::cons(dst->alias_pmt(), dstport));
+
   }
 
   void
diff --git a/gnuradio-runtime/python/gnuradio/gr/qa_hier_block2_message_connections.py b/gnuradio-runtime/python/gnuradio/gr/qa_hier_block2_message_connections.py
new file mode 100644
index 0000000000..bc575b779a
--- /dev/null
+++ b/gnuradio-runtime/python/gnuradio/gr/qa_hier_block2_message_connections.py
@@ -0,0 +1,204 @@
+#!/usr/bin/env python
+#
+# Copyright 2006,2007,2010 Free Software Foundation, Inc.
+#
+# This file is part of GNU Radio
+#
+# GNU Radio is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 3, or (at your option)
+# any later version.
+#
+# GNU Radio is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with GNU Radio; see the file COPYING.  If not, write to
+# the Free Software Foundation, Inc., 51 Franklin Street,
+# Boston, MA 02110-1301, USA.
+#
+
+import weakref
+
+from gnuradio import blocks, gr, gr_unittest
+import pmt
+
+
+class block_with_message_output(gr.basic_block):
+
+    def __init__(self):
+        gr.basic_block.__init__(self,
+            "block_with_message_output",
+            in_sig=None,
+            out_sig=None)
+        self.message_port_register_out(pmt.intern("test"))
+
+
+class block_with_message_input(gr.basic_block):
+
+    def __init__(self):
+        gr.basic_block.__init__(self,
+            "block_with_message_input",
+            in_sig=None,
+            out_sig=None)
+        self.message_port_register_in(pmt.intern("test"))
+
+
+class hier_block_with_message_output(gr.hier_block2):
+
+    def __init__(self):
+        gr.hier_block2.__init__(self,
+            "hier_block_with_message_output",
+            gr.io_signature(0, 0, 0),  # Input signature
+            gr.io_signature(0, 0, 0))  # Output signature
+        self.message_port_register_hier_in("test")
+        self.block = block_with_message_output()
+        self.msg_connect(self.block, "test", weakref.proxy(self), "test")
+
+
+class hier_block_with_message_input(gr.hier_block2):
+
+    def __init__(self):
+        gr.hier_block2.__init__(self,
+            "hier_block_with_message_output",
+            gr.io_signature(0, 0, 0),  # Input signature
+            gr.io_signature(0, 0, 0))  # Output signature
+        self.message_port_register_hier_out("test")
+        self.block = block_with_message_input()
+        self.msg_connect(weakref.proxy(self), "test", self.block, "test")
+
+
+class hier_block_with_message_inout(gr.hier_block2):
+
+    def __init__(self):
+        gr.hier_block2.__init__(self,
+            "hier_block_with_message_inout",
+            gr.io_signature(0, 0, 0),  # Input signature
+            gr.io_signature(0, 0, 0))  # Output signature
+        self.message_port_register_hier_out("test")
+        self.message_port_register_hier_in("test")
+        self.input = block_with_message_input()
+        self.msg_connect(weakref.proxy(self), "test", self.input, "test")
+        self.output = block_with_message_output()
+        self.msg_connect(self.output, "test", weakref.proxy(self), "test")
+
+
+class test_hier_block2_message_connections(gr_unittest.TestCase):
+
+    def setUp(self):
+        self.tb = gr.top_block()
+
+    def tearDown(self):
+        self.tb = None
+
+    def run_top_block(self):
+        self.tb.start()
+        self.tb.stop()
+        self.tb.wait()
+
+    def assert_has_subscription(self, sender, send_port, receiver,
+            receive_port):
+        """assert that the given sender block has a subscription for the given
+        receiver block on the appropriate send and receive ports
+        
+        :param sender: a block sptr to the message sender
+        :param string send_port: the port messages are being sent on
+        :param receiver: a block sptr to the message receiver
+        :param string receive_port: the port messages are being received on
+        """
+        subs = sender.message_subscribers(pmt.intern(send_port))
+        self.assertTrue(pmt.list_has(subs, pmt.cons(
+            pmt.intern(receiver.to_basic_block().alias()),
+            pmt.intern(receive_port))))
+
+    def assert_has_num_subscriptions(self, block, port, number):
+        """assert that the given block has the given number of subscriptions
+        on the given port
+        
+        :param block: a block sptr
+        :param string port: the port name
+        :param number: the number of subscriptions expected
+        """
+        subs = block.message_subscribers(pmt.intern(port))
+        self.assertEqual(pmt.length(subs), number)
+
+    def test_hier_out_to_normal_in(self):
+        message_debug = blocks.message_debug()
+        hier = hier_block_with_message_output()
+
+        self.tb.msg_connect(hier, "test", message_debug, "print")
+        self.run_top_block()
+        self.assert_has_num_subscriptions(hier.block, "test", 1)
+        self.assert_has_num_subscriptions(hier, "test", 0)
+        self.assert_has_subscription(
+            hier.block, "test", message_debug, "print")
+        self.tb.msg_disconnect(hier, "test", message_debug, "print")
+        self.assert_has_num_subscriptions(hier.block, "test", 0)
+
+    def test_normal_out_to_hier_in(self):
+        b = block_with_message_output()
+        hier = hier_block_with_message_input()
+
+        self.tb.msg_connect(b, "test", hier, "test")
+        self.run_top_block()
+        self.assert_has_num_subscriptions(b, "test", 1)
+        self.assert_has_subscription(b, "test", hier.block, "test")
+        self.tb.msg_disconnect(b, "test", hier, "test")
+        self.assert_has_num_subscriptions(b, "test", 0)
+
+    def test_hier_out_to_hier_in(self):
+        hier_out = hier_block_with_message_output()
+        hier_in = hier_block_with_message_input()
+
+        self.tb.msg_connect(hier_out, "test", hier_in, "test")
+        self.run_top_block()
+        self.assert_has_num_subscriptions(hier_out, "test", 0)
+        self.assert_has_num_subscriptions(hier_out.block, "test", 1)
+        self.assert_has_subscription(
+            hier_out.block, "test", hier_in.block, "test")
+        self.tb.msg_disconnect(hier_out, "test", hier_in, "test")
+        self.assert_has_num_subscriptions(hier_out.block, "test", 0)
+
+    def test_normal_in_to_hier_to_normal_out(self):
+        hier = hier_block_with_message_inout()
+        input = block_with_message_output()
+        output = block_with_message_input()
+
+        self.tb.msg_connect(input, "test", hier, "test")
+        self.tb.msg_connect(hier, "test", output, "test")
+        self.run_top_block()
+        self.assert_has_num_subscriptions(input, "test", 1)
+        self.assert_has_subscription(input, "test", hier.input, "test")
+        self.assert_has_num_subscriptions(hier, "test", 0)
+        self.assert_has_num_subscriptions(hier.output, "test", 1)
+        self.assert_has_subscription(hier.output, "test", output, "test")
+        self.tb.msg_disconnect(input, "test", hier, "test")
+        self.tb.msg_disconnect(hier, "test", output, "test")
+        self.assert_has_num_subscriptions(input, "test", 0)
+        self.assert_has_num_subscriptions(hier.output, "test", 0)
+
+    def test_multiple_connections(self):
+        hier = hier_block_with_message_output()
+        x = block_with_message_input()
+        y = block_with_message_input()
+
+        self.tb.msg_connect(hier, "test", x, "test")
+        self.tb.msg_connect(hier, "test", y, "test")
+        self.run_top_block()
+        self.assert_has_num_subscriptions(hier, "test", 0)
+        self.assert_has_num_subscriptions(hier.block, "test", 2)
+        self.assert_has_subscription(hier.block, "test", x, "test")
+        self.assert_has_subscription(hier.block, "test", y, "test")
+        self.tb.msg_disconnect(hier, "test", y, "test")
+        self.assert_has_num_subscriptions(hier, "test", 0)
+        self.assert_has_num_subscriptions(hier.block, "test", 1)
+        self.assert_has_subscription(hier.block, "test", x, "test")
+        self.run_top_block()
+        self.tb.msg_disconnect(hier, "test", x, "test")
+        self.assert_has_num_subscriptions(hier.block, "test", 0)
+
+if __name__ == '__main__':
+    gr_unittest.run(test_hier_block2_message_connections,
+                    "test_hier_block2_message_connections.xml")
author	Mark Cottrell <mark.cottrell@taitradio.com>	2013-10-15 09:05:21 -0700
committer	Johnathan Corgan <johnathan@corganlabs.com>	2013-10-15 09:07:40 -0700
commit	cd19c5a64d14e4cbb90e08648aa45cd76543deed (patch)
tree	f638e92943556229cdf7f1fc450ed0b00bbcbf4b
parent	d5f0cec440ef2848992ce0ae34bfe73bb0c9772a (diff)