From 8b184fda9da4e7fdf08ddfd4d973d5d8d83be308 Mon Sep 17 00:00:00 2001
From: Tom Rondeau <trondeau@vt.edu>
Date: Sat, 6 Nov 2010 14:18:32 -0400
Subject: Adding call in scheduler to handle tag movements between blocks and
some helper functions to get access and keep track of tags.
---
gnuradio-core/src/lib/runtime/gr_block.cc | 7 +++++++
gnuradio-core/src/lib/runtime/gr_block.h | 9 +++++++++
gnuradio-core/src/lib/runtime/gr_block_detail.cc | 19 ++++++++++++++++++-
gnuradio-core/src/lib/runtime/gr_block_detail.h | 9 +++++++++
gnuradio-core/src/lib/runtime/gr_buffer.h | 18 +++++++++++++++++-
gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc | 2 ++
6 files changed, 62 insertions(+), 2 deletions(-)
(limited to 'gnuradio-core/src')
diff --git a/gnuradio-core/src/lib/runtime/gr_block.cc b/gnuradio-core/src/lib/runtime/gr_block.cc
index eb377953d7..51eb5b4986 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block.cc
@@ -166,6 +166,13 @@ gr_block::get_tags_in_range(unsigned int which_output,
return d_detail->get_tags_in_range(which_output, start, end, key);
}
+void
+gr_block::handle_tags()
+{
+ d_detail->handle_tags();
+}
+
+
std::ostream&
operator << (std::ostream& os, const gr_block *m)
{
diff --git a/gnuradio-core/src/lib/runtime/gr_block.h b/gnuradio-core/src/lib/runtime/gr_block.h
index 25886eb103..89d7628472 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_block.h
@@ -209,6 +209,15 @@ class gr_block : public gr_basic_block {
*/
gr_uint64 nitems_written(unsigned int which_output);
+
+ /*!
+ * \brief Function to move tags downstream
+ *
+ * The default behavior proxies to gr_block_detail, which just moves all tags
+ * from input to output and flows them all downstream.
+ */
+ virtual void handle_tags();
+
// ----------------------------------------------------------------------------
private:
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.cc b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
index de4fb21967..7994919d40 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
@@ -39,7 +39,8 @@ gr_block_detail::gr_block_detail (unsigned int ninputs, unsigned int noutputs)
: d_produce_or(0),
d_ninputs (ninputs), d_noutputs (noutputs),
d_input (ninputs), d_output (noutputs),
- d_done (false)
+ d_done (false),
+ d_last_tag(0)
{
s_ncurrently_allocated++;
}
@@ -199,3 +200,19 @@ gr_block_detail::get_tags_in_range(unsigned int which_input,
return found_items_by_key;
}
+
+void
+gr_block_detail::handle_tags()
+{
+ for(unsigned int i = 0; i < d_ninputs; i++) {
+ pmt::pmt_t tuple;
+ while(d_input[i]->get_tag(d_last_tag, tuple)) {
+ d_last_tag++;
+ if(!sink_p()) {
+ for(unsigned int o = 0; o < d_noutputs; o++) {
+ d_output[o]->add_item_tag(tuple);
+ }
+ }
+ }
+ }
+}
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.h b/gnuradio-core/src/lib/runtime/gr_block_detail.h
index ada807f689..cbb59a6893 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.h
@@ -155,6 +155,14 @@ class gr_block_detail {
gr_uint64 abs_end,
const pmt::pmt_t &key);
+ /*!
+ * \brief Default tag handler; moves all tags downstream
+ *
+ * Move all tags from input to output and flows them all downstream. Each input
+ * stream's tags get appended to each output streams tags.
+ */
+ void handle_tags();
+
gr_tpb_detail d_tpb; // used by thread-per-block scheduler
int d_produce_or;
@@ -167,6 +175,7 @@ class gr_block_detail {
std::vector<gr_buffer_sptr> d_output;
bool d_done;
+ size_t d_last_tag; // keep track of which tags we've already received from upstream
gr_block_detail (unsigned int ninputs, unsigned int noutputs);
diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.h b/gnuradio-core/src/lib/runtime/gr_buffer.h
index 8174b7e67a..6dcbff0b90 100644
--- a/gnuradio-core/src/lib/runtime/gr_buffer.h
+++ b/gnuradio-core/src/lib/runtime/gr_buffer.h
@@ -107,6 +107,17 @@ class gr_buffer {
std::deque<pmt::pmt_t>::iterator get_tags_begin() { return d_item_tags.begin(); }
std::deque<pmt::pmt_t>::iterator get_tags_end() { return d_item_tags.end(); }
+ bool get_tag(size_t n, pmt::pmt_t &t)
+ {
+ if(n < d_item_tags.size()) {
+ t = d_item_tags[n];
+ return true;
+ }
+ else {
+ return false;
+ }
+ }
+
// -------------------------------------------------------------------------
private:
@@ -265,7 +276,12 @@ class gr_buffer_reader {
*/
std::deque<pmt::pmt_t> get_tags_in_range(gr_uint64 abs_start,
gr_uint64 abs_end);
-
+
+ bool get_tag(size_t n, pmt::pmt_t &t)
+ {
+ return d_buffer->get_tag(n, t);
+ }
+
// -------------------------------------------------------------------------
private:
diff --git a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
index 03eef17d93..6a84f4be8d 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
@@ -45,6 +45,8 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block)
while ((msg = d->d_tpb.delete_head_nowait()))
block->handle_msg(msg);
+ block->handle_tags();
+
d->d_tpb.clear_changed();
s = d_exec.run_one_iteration();
--
cgit v1.2.3