Merge branch 'maint' of https://github.com/guruofquality/gnuradio into v3.6.4git_gras_support

Conflicts:
	gnuradio-core/src/lib/runtime/gr_block.cc
	gnuradio-core/src/lib/runtime/gr_block.h
	gnuradio-core/src/lib/runtime/gr_types.h

14 files changed, 480 insertions, 36 deletions

diff --git a/gnuradio-core/src/lib/runtime/gr_basic_block.h b/gnuradio-core/src/lib/runtime/gr_basic_block.h
index 9cc2ad775..024159c4c 100644
--- a/gnuradio-core/src/lib/runtime/gr_basic_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_basic_block.h
@@ -54,18 +54,6 @@ class GR_CORE_API gr_basic_block : public gr_msg_accepter, public boost::enable_
   typedef boost::function<void(pmt::pmt_t)> msg_handler_t;
   
  private:
-  /*
-   * This function is called by the runtime system to dispatch messages.
-   *
-   * The thread-safety guarantees mentioned in set_msg_handler are implemented
-   * by the callers of this method.
-   */
-  void dispatch_msg(pmt::pmt_t which_port, pmt::pmt_t msg)
-  {
-    // AA Update this
-    if (d_msg_handlers.find(which_port) != d_msg_handlers.end()) // Is there a handler?
-      d_msg_handlers[which_port](msg); // Yes, invoke it.
-  };
   
   //msg_handler_t	 d_msg_handler;
   typedef std::map<pmt::pmt_t , msg_handler_t, pmt::pmt_comperator> d_msg_handlers_t;
@@ -117,6 +105,27 @@ class GR_CORE_API gr_basic_block : public gr_msg_accepter, public boost::enable_
    */
   void set_color(vcolor color) { d_color = color; }
   vcolor color() const { return d_color; }
+
+  /*!
+   * \brief Tests if there is a handler attached to port \p which_port
+   */
+   bool has_msg_handler(pmt::pmt_t which_port) {
+     return (d_msg_handlers.find(which_port) != d_msg_handlers.end());
+   }
+
+  /*
+   * This function is called by the runtime system to dispatch messages.
+   *
+   * The thread-safety guarantees mentioned in set_msg_handler are implemented
+   * by the callers of this method.
+   */
+  virtual void dispatch_msg(pmt::pmt_t which_port, pmt::pmt_t msg)
+  {
+    // AA Update this
+    if(has_msg_handler(which_port)) {  // Is there a handler?
+      d_msg_handlers[which_port](msg); // Yes, invoke it.
+    }
+  }
   
   // Message passing interface
   pmt::pmt_t message_subscribers;
@@ -176,9 +185,18 @@ class GR_CORE_API gr_basic_block : public gr_msg_accepter, public boost::enable_
   }
   bool empty_p() { 
     bool rv = true;
-    BOOST_FOREACH(msg_queue_map_t::value_type &i, msg_queue){ rv &= msg_queue[i.first].empty(); }
+    BOOST_FOREACH(msg_queue_map_t::value_type &i, msg_queue) {
+      rv &= msg_queue[i.first].empty();
+    }
     return rv;
   }
+
+  //! How many messages in the queue?
+  size_t nmsgs(pmt::pmt_t which_port) { 
+    if(msg_queue.find(which_port) == msg_queue.end())
+      throw std::runtime_error("port does not exist!");
+    return msg_queue[which_port].size(); 
+  }
   
   //| Acquires and release the mutex
   void insert_tail( pmt::pmt_t which_port, pmt::pmt_t msg);
diff --git a/gnuradio-core/src/lib/runtime/gr_block.h b/gnuradio-core/src/lib/runtime/gr_block.h
index 6c6480eb4..9c8485663 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_block.h
@@ -27,10 +27,6 @@
 #include <map>
 #include <boost/foreach.hpp>
 
-typedef std::vector<int> gr_vector_int;
-typedef std::vector<void *> gr_vector_void_star;
-typedef std::vector<const void *> gr_vector_const_void_star;
-
 namespace gnuradio
 {
 //! dummy entry, just here for legacy purposes
@@ -227,6 +223,10 @@ struct GR_CORE_API gr_block : gras::Block
     msg_queue_map_t msg_queue;
     pmt::pmt_t message_subscribers;
 
+  typedef boost::function<void(pmt::pmt_t)> msg_handler_t;
+  typedef std::map<pmt::pmt_t , msg_handler_t, pmt::pmt_comperator> d_msg_handlers_t;
+  d_msg_handlers_t d_msg_handlers;
+
     template <typename T> void set_msg_handler(pmt::pmt_t which_port, T msg_handler){}
 
     void message_port_register_in(pmt::pmt_t /*port_id*/){}
@@ -297,6 +297,27 @@ struct GR_CORE_API gr_block : gras::Block
         return false;
     }
 
+  /*!
+   * \brief Tests if there is a handler attached to port \p which_port
+   */
+   bool has_msg_handler(pmt::pmt_t which_port) {
+     return (d_msg_handlers.find(which_port) != d_msg_handlers.end());
+   }
+
+  /*
+   * This function is called by the runtime system to dispatch messages.
+   *
+   * The thread-safety guarantees mentioned in set_msg_handler are implemented
+   * by the callers of this method.
+   */
+  virtual void dispatch_msg(pmt::pmt_t which_port, pmt::pmt_t msg)
+  {
+    // AA Update this
+    if(has_msg_handler(which_port)) {  // Is there a handler?
+      d_msg_handlers[which_port](msg); // Yes, invoke it.
+    }
+  }
+
     ///////////////// private vars //////////////////////
 
     gr_vector_int _work_ninput_items;
diff --git a/gnuradio-core/src/lib/runtime/gr_block.i b/gnuradio-core/src/lib/runtime/gr_block.i
index db6c1d04a..a80f64d02 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.i
+++ b/gnuradio-core/src/lib/runtime/gr_block.i
@@ -66,6 +66,27 @@ class gr_block : public gr_basic_block {
   void set_min_output_buffer(long min_output_buffer);
   void set_min_output_buffer(int port, long min_output_buffer);
 
+  // Methods to access performance counters
+  float pc_noutput_items();
+  float pc_noutput_items_var();
+  float pc_nproduced();
+  float pc_nproduced_var();
+  float pc_input_buffers_full(int which);
+  float pc_input_buffers_full_var(int which);
+  std::vector<float> pc_input_buffers_full();
+  std::vector<float> pc_input_buffers_full_var();
+  float pc_output_buffers_full(int which);
+  float pc_output_buffers_full_var(int which);
+  std::vector<float> pc_output_buffers_full();
+  std::vector<float> pc_output_buffers_full_var();
+  float pc_work_time();
+  float pc_work_time_var();
+  
+  // Methods to manage processor affinity.
+  void set_processor_affinity(const gr_vector_uint &mask);
+  void unset_processor_affinity();
+  gr_vector_uint processor_affinity();
+
   // internal use
   gr_block_detail_sptr detail () const { return d_detail; }
   void set_detail (gr_block_detail_sptr detail) { d_detail = detail; }
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.cc b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
index 337c9518e..82081039a 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.cc
@@ -26,6 +26,7 @@
 
 #include <gr_block_detail.h>
 #include <gr_buffer.h>
+#include <iostream>
 
 using namespace pmt;
 
@@ -41,7 +42,18 @@ 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_avg_noutput_items(0), 
+    d_var_noutput_items(0),
+    d_avg_nproduced(0),
+    d_var_nproduced(0),
+    d_avg_input_buffers_full(ninputs, 0),
+    d_var_input_buffers_full(ninputs, 0),
+    d_avg_output_buffers_full(noutputs, 0),
+    d_var_output_buffers_full(noutputs, 0),
+    d_avg_work_time(0),
+    d_var_work_time(0),
+    d_pc_counter(0)
 {
   s_ncurrently_allocated++;
 }
@@ -156,6 +168,18 @@ gr_block_detail::add_item_tag(unsigned int which_output, const gr_tag_t &tag)
 }
 
 void
+gr_block_detail::remove_item_tag(unsigned int which_input, const gr_tag_t &tag)
+{
+  if(!pmt_is_symbol(tag.key)) {
+    throw pmt_wrong_type("gr_block_detail::add_item_tag key", tag.key);
+  }
+  else {
+    // Add tag to gr_buffer's deque tags
+    d_input[which_input]->buffer()->remove_item_tag(tag);
+  }
+}
+
+void
 gr_block_detail::get_tags_in_range(std::vector<gr_tag_t> &v,
 				   unsigned int which_input,
 				   uint64_t abs_start,
@@ -189,3 +213,200 @@ gr_block_detail::get_tags_in_range(std::vector<gr_tag_t> &v,
     }
   }
 }
+
+void
+gr_block_detail::set_processor_affinity(const std::vector<unsigned int> &mask)
+{
+  if(threaded) {
+    try {
+      gruel::thread_bind_to_processor(thread, mask);
+    }
+    catch (std::runtime_error e) {
+      std::cerr << "set_processor_affinity: invalid mask."  << std::endl;;
+    }
+  }
+}
+
+void
+gr_block_detail::unset_processor_affinity()
+{
+  if(threaded) {
+    gruel::thread_unbind(thread);
+  }
+}
+
+void
+gr_block_detail::start_perf_counters()
+{
+  d_start_of_work = gruel::high_res_timer_now();
+}
+
+void
+gr_block_detail::stop_perf_counters(int noutput_items, int nproduced)
+{
+  d_end_of_work = gruel::high_res_timer_now();
+  gruel::high_res_timer_type diff = d_end_of_work - d_start_of_work;
+
+  if(d_pc_counter == 0) {
+    d_avg_work_time = diff;
+    d_var_work_time = 0;
+    d_avg_nproduced = nproduced;
+    d_var_nproduced = 0;
+    d_avg_noutput_items = noutput_items;
+    d_var_noutput_items = 0;
+    for(size_t i=0; i < d_input.size(); i++) {
+      float pfull = static_cast<float>(d_input[i]->items_available()) /
+	static_cast<float>(d_input[i]->max_possible_items_available());
+      d_avg_input_buffers_full[i] = pfull;
+      d_var_input_buffers_full[i] = 0;
+    }
+    for(size_t i=0; i < d_output.size(); i++) {
+      float pfull = 1.0f - static_cast<float>(d_output[i]->space_available()) /
+	static_cast<float>(d_output[i]->bufsize());
+      d_avg_output_buffers_full[i] = pfull;
+      d_var_output_buffers_full[i] = 0;
+    }
+  }
+  else {
+    float d = diff - d_avg_work_time;
+    d_avg_work_time = d_avg_work_time + d/d_pc_counter;
+    d_var_work_time = d_var_work_time + d*d;
+
+    d = nproduced - d_avg_nproduced;
+    d_avg_nproduced = d_avg_nproduced +	d/d_pc_counter;
+    d_var_nproduced = d_var_nproduced +	d*d;
+
+    d = noutput_items - d_avg_noutput_items;
+    d_avg_noutput_items = d_avg_noutput_items + d/d_pc_counter;
+    d_var_noutput_items = d_var_noutput_items + d*d;
+
+    for(size_t i=0; i < d_input.size(); i++) {
+      float pfull = static_cast<float>(d_input[i]->items_available()) /
+	static_cast<float>(d_input[i]->max_possible_items_available());
+      
+      d = pfull - d_avg_input_buffers_full[i];
+      d_avg_input_buffers_full[i] = d_avg_input_buffers_full[i] + d/d_pc_counter;
+      d_var_input_buffers_full[i] = d_var_input_buffers_full[i] + d*d;
+    }
+
+    for(size_t i=0; i < d_output.size(); i++) {
+      float pfull = 1.0f - static_cast<float>(d_output[i]->space_available()) /
+	static_cast<float>(d_output[i]->bufsize());
+
+      d = pfull - d_avg_output_buffers_full[i];
+      d_avg_output_buffers_full[i] = d_avg_output_buffers_full[i] + d/d_pc_counter;
+      d_var_output_buffers_full[i] = d_var_output_buffers_full[i] + d*d;
+    }
+  }
+
+  d_pc_counter++;
+}
+
+void
+gr_block_detail::reset_perf_counters()
+{
+  d_pc_counter = 0;
+}
+
+float
+gr_block_detail::pc_noutput_items()
+{
+  return d_avg_noutput_items;
+}
+
+float
+gr_block_detail::pc_nproduced()
+{
+  return d_avg_nproduced;
+}
+
+float
+gr_block_detail::pc_input_buffers_full(size_t which)
+{
+  if(which < d_avg_input_buffers_full.size())
+    return d_avg_input_buffers_full[which];
+  else
+    return 0;
+}
+
+std::vector<float>
+gr_block_detail::pc_input_buffers_full()
+{
+  return d_avg_input_buffers_full;
+}
+
+float
+gr_block_detail::pc_output_buffers_full(size_t which)
+{
+  if(which < d_avg_output_buffers_full.size())
+    return d_avg_output_buffers_full[which];
+  else
+    return 0;
+}
+
+std::vector<float>
+gr_block_detail::pc_output_buffers_full()
+{
+  return d_avg_output_buffers_full;
+}
+
+float
+gr_block_detail::pc_work_time()
+{
+  return d_avg_work_time;
+}
+
+
+float
+gr_block_detail::pc_noutput_items_var()
+{
+  return d_var_noutput_items/(d_pc_counter-1);
+}
+
+float
+gr_block_detail::pc_nproduced_var()
+{
+  return d_var_nproduced/(d_pc_counter-1);
+}
+
+float
+gr_block_detail::pc_input_buffers_full_var(size_t which)
+{
+  if(which < d_avg_input_buffers_full.size())
+    return d_var_input_buffers_full[which]/(d_pc_counter-1);
+  else
+    return 0;
+}
+
+std::vector<float>
+gr_block_detail::pc_input_buffers_full_var()
+{
+  std::vector<float> var(d_avg_input_buffers_full.size(), 0);
+  for(size_t i = 0; i < d_avg_input_buffers_full.size(); i++)
+    var[i] = d_avg_input_buffers_full[i]/(d_pc_counter-1);
+  return var;
+}
+
+float
+gr_block_detail::pc_output_buffers_full_var(size_t which)
+{
+  if(which < d_avg_output_buffers_full.size())
+    return d_var_output_buffers_full[which]/(d_pc_counter-1);
+  else
+    return 0;
+}
+
+std::vector<float>
+gr_block_detail::pc_output_buffers_full_var()
+{
+  std::vector<float> var(d_avg_output_buffers_full.size(), 0);
+  for(size_t i = 0; i < d_avg_output_buffers_full.size(); i++)
+    var[i] = d_avg_output_buffers_full[i]/(d_pc_counter-1);
+  return var;
+}
+
+float
+gr_block_detail::pc_work_time_var()
+{
+  return d_var_work_time/(d_pc_counter-1);
+}
diff --git a/gnuradio-core/src/lib/runtime/gr_block_detail.h b/gnuradio-core/src/lib/runtime/gr_block_detail.h
index 16d9f4d42..32a01e763 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_detail.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_detail.h
@@ -27,6 +27,7 @@
 #include <gr_runtime_types.h>
 #include <gr_tpb_detail.h>
 #include <gr_tags.h>
+#include <gruel/high_res_timer.h>
 #include <stdexcept>
 
 /*!
@@ -95,8 +96,7 @@ class GR_CORE_API gr_block_detail {
   /*!
    * \brief  Adds a new tag to the given output stream.
    *
-   * This takes the input parameters and builds a PMT tuple
-   * from it. It then calls gr_buffer::add_item_tag(pmt::pmt_t t),
+   * Calls gr_buffer::add_item_tag(),
    * which appends the tag onto its deque.
    *
    * \param which_output  an integer of which output stream to attach the tag
@@ -105,6 +105,16 @@ class GR_CORE_API gr_block_detail {
   void add_item_tag(unsigned int which_output, const gr_tag_t &tag);
 
   /*!
+   * \brief  Removes a tag from the given input stream.
+   *
+   * Calls gr_buffer::remove_item_tag(), which removes the tag from its deque.
+   *
+   * \param which_input  an integer of which input stream to remove the tag from
+   * \param tag the tag object to add
+   */
+  void remove_item_tag(unsigned int which_input, const gr_tag_t &tag);
+
+  /*!
    * \brief Given a [start,end), returns a vector of all tags in the range.
    *
    * Pass-through function to gr_buffer_reader to get a vector of tags
@@ -146,6 +156,42 @@ class GR_CORE_API gr_block_detail {
 			 uint64_t abs_end,
 			 const pmt::pmt_t &key);
 
+  /*!
+   * \brief Set core affinity of block to the cores in the vector mask.
+   *
+   * \param mask a vector of unsigned ints of the core numbers available to this block.
+   */
+  void set_processor_affinity(const std::vector<unsigned int> &mask);
+
+  /*!
+   * \brief Unset core affinity.
+   */
+  void unset_processor_affinity();
+
+  bool                               threaded;  // set if thread is currently running.
+  gruel::gr_thread_t                 thread;    // portable thread handle
+
+  void start_perf_counters();
+  void stop_perf_counters(int noutput_items, int nproduced);
+  void reset_perf_counters();
+
+  // Calls to get performance counter items
+  float pc_noutput_items();
+  float pc_nproduced();
+  float pc_input_buffers_full(size_t which);
+  std::vector<float> pc_input_buffers_full();
+  float pc_output_buffers_full(size_t which);
+  std::vector<float> pc_output_buffers_full();
+  float pc_work_time();
+
+  float pc_noutput_items_var();
+  float pc_nproduced_var();
+  float pc_input_buffers_full_var(size_t which);
+  std::vector<float> pc_input_buffers_full_var();
+  float pc_output_buffers_full_var(size_t which);
+  std::vector<float> pc_output_buffers_full_var();
+  float pc_work_time_var();
+ 
   gr_tpb_detail			     d_tpb;	// used by thread-per-block scheduler
   int				     d_produce_or;
 
@@ -158,6 +204,20 @@ class GR_CORE_API gr_block_detail {
   std::vector<gr_buffer_sptr>	     d_output;
   bool                               d_done;
 
+  // Performance counters
+  float d_avg_noutput_items;
+  float d_var_noutput_items;
+  float d_avg_nproduced;
+  float d_var_nproduced;
+  std::vector<float> d_avg_input_buffers_full;
+  std::vector<float> d_var_input_buffers_full;
+  std::vector<float> d_avg_output_buffers_full;
+  std::vector<float> d_var_output_buffers_full;
+  gruel::high_res_timer_type d_start_of_work, d_end_of_work;
+  float d_avg_work_time;
+  float d_var_work_time;
+  float d_pc_counter;
+  
   gr_block_detail (unsigned int ninputs, unsigned int noutputs);
 
   friend struct gr_tpb_detail;
diff --git a/gnuradio-core/src/lib/runtime/gr_block_executor.cc b/gnuradio-core/src/lib/runtime/gr_block_executor.cc
index 375b58f56..e070f3c50 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_executor.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_executor.cc
@@ -28,6 +28,7 @@
 #include <gr_block.h>
 #include <gr_block_detail.h>
 #include <gr_buffer.h>
+#include <gr_prefs.h>
 #include <boost/thread.hpp>
 #include <boost/format.hpp>
 #include <iostream>
@@ -64,22 +65,21 @@ round_down (unsigned int n, unsigned int multiple)
 // on is done.
 //
 static int
-min_available_space (gr_block_detail *d, int output_multiple)
+min_available_space (gr_block_detail *d, int output_multiple, int min_noutput_items)
 {
-  int	min_space = std::numeric_limits<int>::max();
-
+  int min_space = std::numeric_limits<int>::max();
+  if (min_noutput_items == 0)
+    min_noutput_items = 1;
   for (int i = 0; i < d->noutputs (); i++){
     gruel::scoped_lock guard(*d->output(i)->mutex());
-#if 0
-    int n = round_down(d->output(i)->space_available(), output_multiple);
-#else
-    int n = round_down(std::min(d->output(i)->space_available(),
-				d->output(i)->bufsize()/2),
-		       output_multiple);
-#endif
-    if (n == 0){			// We're blocked on output.
-      if (d->output(i)->done()){	// Downstream is done, therefore we're done.
-	return -1;
+    int avail_n = round_down(d->output(i)->space_available(), output_multiple);
+    int best_n = round_down(d->output(i)->bufsize()/2, output_multiple);
+    if (best_n < min_noutput_items)
+      throw std::runtime_error("Buffer too small for min_noutput_items");
+    int n = std::min(avail_n, best_n);
+    if (n < min_noutput_items){  // We're blocked on output.
+      if (d->output(i)->done()){ // Downstream is done, therefore we're done.
+        return -1;
       }
       return 0;
     }
@@ -166,6 +166,11 @@ gr_block_executor::gr_block_executor (gr_block_sptr block, int max_noutput_items
 	   << d_block << std::endl;
   }
 
+#ifdef GR_PERFORMANCE_COUNTERS
+  gr_prefs *prefs = gr_prefs::singleton();
+  d_use_pc = prefs->get_bool("PerfCounters", "on", false);
+#endif /* GR_PERFORMANCE_COUNTERS */
+
   d_block->start();			// enable any drivers, etc.
 }
 
@@ -205,7 +210,7 @@ gr_block_executor::run_one_iteration()
     d_start_nitems_read.resize(0);
 
     // determine the minimum available output space
-    noutput_items = min_available_space (d, m->output_multiple ());
+    noutput_items = min_available_space (d, m->output_multiple (), m->min_noutput_items ());
     noutput_items = std::min(noutput_items, max_noutput_items);
     LOG(*d_log << " source\n  noutput_items = " << noutput_items << std::endl);
     if (noutput_items == -1)		// we're done
@@ -286,7 +291,7 @@ gr_block_executor::run_one_iteration()
     }
 
     // determine the minimum available output space
-    noutput_items = min_available_space (d, m->output_multiple ());
+    noutput_items = min_available_space (d, m->output_multiple (), m->min_noutput_items ());
     if (ENABLE_LOGGING){
       *d_log << " regular ";
       if (m->relative_rate() >= 1.0)
@@ -420,9 +425,20 @@ gr_block_executor::run_one_iteration()
     for (int i = 0; i < d->ninputs(); i++)
       d_start_nitems_read[i] = d->nitems_read(i);
 
+#ifdef GR_PERFORMANCE_COUNTERS
+    if(d_use_pc)
+      d->start_perf_counters();
+#endif /* GR_PERFORMANCE_COUNTERS */
+    
     // Do the actual work of the block
     int n = m->general_work (noutput_items, d_ninput_items,
 			     d_input_items, d_output_items);
+
+#ifdef GR_PERFORMANCE_COUNTERS
+    if(d_use_pc)
+      d->stop_perf_counters(noutput_items, n);
+#endif /* GR_PERFORMANCE_COUNTERS */
+
     LOG(*d_log << "  general_work: noutput_items = " << noutput_items
 	<< " result = " << n << std::endl);
 
diff --git a/gnuradio-core/src/lib/runtime/gr_block_executor.h b/gnuradio-core/src/lib/runtime/gr_block_executor.h
index 0ae5affba..fb7f9c269 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_executor.h
+++ b/gnuradio-core/src/lib/runtime/gr_block_executor.h
@@ -53,6 +53,10 @@ protected:
   std::vector<gr_tag_t>       d_returned_tags;
   int                           d_max_noutput_items;
 
+#ifdef GR_PERFORMANCE_COUNTERS
+  bool d_use_pc;
+#endif /* GR_PERFORMANCE_COUNTERS */
+
  public:
   gr_block_executor(gr_block_sptr block, int max_noutput_items=100000);
   ~gr_block_executor ();
diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.cc b/gnuradio-core/src/lib/runtime/gr_buffer.cc
index b923ca57a..369959d65 100644
--- a/gnuradio-core/src/lib/runtime/gr_buffer.cc
+++ b/gnuradio-core/src/lib/runtime/gr_buffer.cc
@@ -234,6 +234,18 @@ gr_buffer::add_item_tag(const gr_tag_t &tag)
 }
 
 void
+gr_buffer::remove_item_tag(const gr_tag_t &tag)
+{
+  gruel::scoped_lock guard(*mutex());
+  for (std::deque<gr_tag_t>::iterator it = d_item_tags.begin(); it != d_item_tags.end(); ++it) {
+    if (*it == tag) {
+      d_item_tags.erase(it);
+      break;
+    }
+  }
+}
+
+void
 gr_buffer::prune_tags(uint64_t max_time)
 {
   /* NOTE: this function _should_ lock the mutex before editing
diff --git a/gnuradio-core/src/lib/runtime/gr_buffer.h b/gnuradio-core/src/lib/runtime/gr_buffer.h
index 67d48fb2d..28ea97726 100644
--- a/gnuradio-core/src/lib/runtime/gr_buffer.h
+++ b/gnuradio-core/src/lib/runtime/gr_buffer.h
@@ -103,6 +103,15 @@ class GR_CORE_API gr_buffer {
   void add_item_tag(const gr_tag_t &tag);
 
   /*!
+   * \brief  Removes an existing tag from the buffer.
+   *
+   * If no such tag is found, does nothing.
+   *
+   * \param tag        the tag that needs to be removed
+   */
+  void remove_item_tag(const gr_tag_t &tag);
+
+  /*!
    * \brief  Removes all tags before \p max_time from buffer
    *
    * \param max_time        the time (item number) to trim up until.
diff --git a/gnuradio-core/src/lib/runtime/gr_tags.h b/gnuradio-core/src/lib/runtime/gr_tags.h
index 8bffcd0fe..a9ca90235 100644
--- a/gnuradio-core/src/lib/runtime/gr_tags.h
+++ b/gnuradio-core/src/lib/runtime/gr_tags.h
@@ -45,6 +45,11 @@ struct GR_CORE_API gr_tag_t{
     ){
         return x.offset < y.offset;
     }
+
+    inline bool operator == (const gr_tag_t &t) const
+    {
+      return (t.key == key) && (t.value == value) && (t.srcid == srcid) && (t.offset == offset);
+    }
 };
 
 #endif /*INCLUDED_GR_TAGS_H*/
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 9f17a48a8..679fd1512 100644
--- a/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
+++ b/gnuradio-core/src/lib/runtime/gr_tpb_thread_body.cc
@@ -22,6 +22,7 @@
 #include <config.h>
 #endif
 #include <gr_tpb_thread_body.h>
+#include <gr_prefs.h>
 #include <iostream>
 #include <boost/thread.hpp>
 #include <gruel/pmt.h>
@@ -38,6 +39,16 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
   gr_block_executor::state s;
   pmt_t msg;
 
+  d->threaded = true;
+  d->thread = gruel::get_current_thread_id();
+
+  gr_prefs *p = gr_prefs::singleton();
+  size_t max_nmsgs = static_cast<size_t>(p->get_long("DEFAULT", "max_messages", 100));
+
+  // Set thread affinity if it was set before fg was started.
+  if(block->processor_affinity().size() > 0) {
+    gruel::thread_bind_to_processor(d->thread, block->processor_affinity());
+  }
 
   while (1){
     boost::this_thread::interruption_point();
@@ -47,9 +58,20 @@ gr_tpb_thread_body::gr_tpb_thread_body(gr_block_sptr block, int max_noutput_item
     
     BOOST_FOREACH( gr_basic_block::msg_queue_map_t::value_type &i, block->msg_queue )
     {
+      // Check if we have a message handler attached before getting
+      // any messages. This is mostly a protection for the unknown
+      // startup sequence of the threads.
+      if(block->has_msg_handler(i.first)) {
         while ((msg = block->delete_head_nowait(i.first))){
-            block->dispatch_msg(i.first,msg);
+          block->dispatch_msg(i.first,msg);
         }
+      }
+      else {
+        // If we don't have a handler but are building up messages,
+        // prune the queue from the front to keep memory in check.
+        if(block->nmsgs(i.first) > max_nmsgs)
+          msg = block->delete_head_nowait(i.first);
+      }
     }
 
     d->d_tpb.clear_changed();
diff --git a/gnuradio-core/src/lib/runtime/gr_types.h b/gnuradio-core/src/lib/runtime/gr_types.h
index a698226bf..9b70eb20c 100644
--- a/gnuradio-core/src/lib/runtime/gr_types.h
+++ b/gnuradio-core/src/lib/runtime/gr_types.h
@@ -35,6 +35,13 @@ typedef boost::uint16_t gr_uint16;
 typedef boost::uint32_t gr_uint32;
 typedef boost::uint64_t gr_uint64;
 
+typedef std::vector<int>			gr_vector_int;
+typedef std::vector<unsigned int>		gr_vector_uint;
+typedef std::vector<float>			gr_vector_float;
+typedef std::vector<double>			gr_vector_double;
+typedef std::vector<void *>			gr_vector_void_star;
+typedef std::vector<const void *>		gr_vector_const_void_star;
+
 #include <complex>
 typedef std::complex<float> gr_complex;
 typedef std::complex<double> gr_complexd;
diff --git a/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc b/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc
index a0b4755a8..1d3dafadf 100644
--- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc
+++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc
@@ -262,3 +262,24 @@ void qa_gr_top_block::t10_reconfig_max_output_buffer()
   // Wait for flowgraph to end on its own
   tb->wait();
 }
+
+void qa_gr_top_block::t11_set_block_affinity()
+{
+  gr_top_block_sptr tb = gr_make_top_block("top");
+  gr_block_sptr src (gr_make_null_source(sizeof(float)));
+  gr_block_sptr snk (gr_make_null_sink(sizeof(float)));
+
+  std::vector<unsigned int> set(1, 0), ret;
+  src->set_processor_affinity(set);
+
+  tb->connect(src, 0, snk, 0);
+  tb->start();
+  tb->stop();
+  tb->wait();
+
+  ret = src->processor_affinity();
+
+  // We only set the core affinity to 0 because we always know at
+  // least one thread core exists to use.
+  CPPUNIT_ASSERT_EQUAL(set[0], ret[0]);
+}
diff --git a/gnuradio-core/src/lib/runtime/qa_gr_top_block.h b/gnuradio-core/src/lib/runtime/qa_gr_top_block.h
index bb891abca..634eeab1f 100644
--- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.h
+++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.h
@@ -38,6 +38,11 @@ class qa_gr_top_block : public CppUnit::TestCase
   CPPUNIT_TEST(t4_reconfigure);  // triggers 'join never returns' bug
   CPPUNIT_TEST(t5_max_noutputs);
   CPPUNIT_TEST(t6_reconfig_max_noutputs);
+  CPPUNIT_TEST(t7_max_noutputs_per_block);
+  CPPUNIT_TEST(t8_reconfig_max_noutputs_per_block);
+  CPPUNIT_TEST(t9_max_output_buffer);
+  CPPUNIT_TEST(t10_reconfig_max_output_buffer);
+  CPPUNIT_TEST(t11_set_block_affinity);
 
   CPPUNIT_TEST_SUITE_END();
 
@@ -54,6 +59,8 @@ private:
   void t8_reconfig_max_noutputs_per_block();
   void t9_max_output_buffer();
   void t10_reconfig_max_output_buffer();
+  void t11_set_block_affinity();
+
 };
 
 #endif /* INCLUDED_QA_GR_TOP_BLOCK_H */