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

41 files changed, 1557 insertions, 79 deletions

diff --git a/gnuradio-core/src/examples/mp-sched/CMakeLists.txt b/gnuradio-core/src/examples/mp-sched/CMakeLists.txt
index dc47d17f9..d2d910ecf 100644
--- a/gnuradio-core/src/examples/mp-sched/CMakeLists.txt
+++ b/gnuradio-core/src/examples/mp-sched/CMakeLists.txt
@@ -20,6 +20,7 @@
 include(GrPython)
 
 GR_PYTHON_INSTALL(PROGRAMS
+  affinity_set.py
   plot_flops.py
   run_synthetic.py
   synthetic.py
diff --git a/gnuradio-core/src/examples/mp-sched/affinity_set.py b/gnuradio-core/src/examples/mp-sched/affinity_set.py
new file mode 100755
index 000000000..6db632e0f
--- /dev/null
+++ b/gnuradio-core/src/examples/mp-sched/affinity_set.py
@@ -0,0 +1,73 @@
+#!/usr/bin/env python
+##################################################
+# Gnuradio Python Flow Graph
+# Title: Affinity Set Test
+##################################################
+
+from gnuradio import eng_notation
+from gnuradio import gr
+from gnuradio.eng_option import eng_option
+from gnuradio.gr import firdes
+from optparse import OptionParser
+import sys
+
+class affinity_set(gr.top_block):
+
+    def __init__(self):
+        gr.top_block.__init__(self, "Affinity Set Test")
+
+        ##################################################
+	# Variables
+	##################################################
+	self.samp_rate = samp_rate = 32000
+
+	##################################################
+	# Blocks
+	##################################################
+        vec_len = 1
+	self.gr_throttle_0 = gr.throttle(gr.sizeof_gr_complex*vec_len, samp_rate)
+	self.gr_null_source_0 = gr.null_source(gr.sizeof_gr_complex*vec_len)
+	self.gr_null_sink_0 = gr.null_sink(gr.sizeof_gr_complex*vec_len)
+	self.gr_filt_0 = gr.fir_filter_ccc(1, 40000*[0.2+0.3j,])
+	self.gr_filt_1 = gr.fir_filter_ccc(1, 40000*[0.2+0.3j,])
+
+	self.gr_filt_0.set_processor_affinity([0,])
+	self.gr_filt_1.set_processor_affinity([0,1])
+		
+	##################################################
+	# Connections
+	##################################################
+	self.connect((self.gr_null_source_0, 0), (self.gr_throttle_0, 0))
+	self.connect((self.gr_throttle_0, 0), (self.gr_filt_0, 0))
+	self.connect((self.gr_filt_0, 0), (self.gr_filt_1, 0))
+	self.connect((self.gr_filt_1, 0), (self.gr_null_sink_0, 0))
+
+
+	# QT sink close method reimplementation
+
+    def get_samp_rate(self):
+        return self.samp_rate
+
+    def set_samp_rate(self, samp_rate):
+        self.samp_rate = samp_rate
+
+if __name__ == '__main__':
+    parser = OptionParser(option_class=eng_option, usage="%prog: [options]")
+    (options, args) = parser.parse_args()
+    tb = affinity_set()
+    tb.start()
+	
+    while(1):
+        ret = raw_input('Press Enter to quit: ')
+	if(len(ret) == 0):
+            tb.stop()
+	    sys.exit(0)
+        elif(ret.lower() == "none"):
+            tb.gr_filt_0.unset_processor_affinity()
+	else:
+            try:
+                n = int(ret)
+	    except ValueError:
+                print "Invalid number"
+	    else:
+                tb.gr_filt_0.set_processor_affinity([n,])
diff --git a/gnuradio-core/src/lib/general/CMakeLists.txt b/gnuradio-core/src/lib/general/CMakeLists.txt
index 4c99acfc3..ce9a80f37 100644
--- a/gnuradio-core/src/lib/general/CMakeLists.txt
+++ b/gnuradio-core/src/lib/general/CMakeLists.txt
@@ -259,6 +259,7 @@ set(gr_core_general_triple_threats
     gr_pwr_squelch_cc
     gr_pwr_squelch_ff
     gr_quadrature_demod_cf
+    gr_random_pdu
     gr_rail_ff
     gr_regenerate_bb
     gr_remez
diff --git a/gnuradio-core/src/lib/general/general.i b/gnuradio-core/src/lib/general/general.i
index 1446088a2..b727bc8a6 100644
--- a/gnuradio-core/src/lib/general/general.i
+++ b/gnuradio-core/src/lib/general/general.i
@@ -70,6 +70,7 @@
 #include <gr_agc_cc.h>
 #include <gr_agc2_ff.h>
 #include <gr_agc2_cc.h>
+#include <gr_random_pdu.h>
 #include <gr_rms_cf.h>
 #include <gr_rms_ff.h>
 #include <gr_nlog10_ff.h>
@@ -194,6 +195,7 @@
 %include "gr_agc_cc.i"
 %include "gr_agc2_ff.i"
 %include "gr_agc2_cc.i"
+%include "gr_random_pdu.i"
 %include "gr_rms_cf.i"
 %include "gr_rms_ff.i"
 %include "gr_nlog10_ff.i"
diff --git a/gnuradio-core/src/lib/general/gr_block_gateway.h b/gnuradio-core/src/lib/general/gr_block_gateway.h
index ae91d41b5..2452b1045 100644
--- a/gnuradio-core/src/lib/general/gr_block_gateway.h
+++ b/gnuradio-core/src/lib/general/gr_block_gateway.h
@@ -188,6 +188,58 @@ public:
         gr_block::get_tags_in_range(tags, which_input, abs_start, abs_end, key);
         return tags;
     }
+
+    /* Message passing interface */
+    void gr_block__message_port_register_in(pmt::pmt_t port_id){
+        gr_block::message_port_register_in(port_id);
+    }
+
+    void gr_block__message_port_register_out(pmt::pmt_t port_id){
+        gr_block::message_port_register_out(port_id);
+    }
+
+    void gr_block__message_port_pub(pmt::pmt_t port_id, pmt::pmt_t msg){
+        gr_block::message_port_pub(port_id, msg);
+    }
+
+    void gr_block__message_port_sub(pmt::pmt_t port_id, pmt::pmt_t target){
+        gr_block::message_port_sub(port_id, target);
+    }
+
+    void gr_block__message_port_unsub(pmt::pmt_t port_id, pmt::pmt_t target){
+        gr_block::message_port_unsub(port_id, target);
+    }
+    
+    pmt::pmt_t gr_block__message_ports_in(){
+        return gr_block::message_ports_in();
+    }
+
+    pmt::pmt_t gr_block__message_ports_out(){
+        return gr_block::message_ports_out();
+    }
+
+    void set_msg_handler_feval(pmt::pmt_t which_port, gr_feval_p *msg_handler)
+    {
+        if(msg_queue.find(which_port) == msg_queue.end()){ 
+	    throw std::runtime_error("attempt to set_msg_handler_feval() on bad input message port!"); 
+	}
+	d_msg_handlers_feval[which_port] = msg_handler;
+    }
+
+protected:
+    typedef std::map<pmt::pmt_t, gr_feval_p *, pmt::pmt_comperator> msg_handlers_feval_t;
+    msg_handlers_feval_t d_msg_handlers_feval;
+
+    void dispatch_msg(pmt::pmt_t which_port, pmt::pmt_t msg){
+        // Is there a handler?
+        if (d_msg_handlers_feval.find(which_port) != d_msg_handlers_feval.end()){
+    	    d_msg_handlers_feval[which_port]->calleval(msg); // Yes, invoke it.
+        }
+        else {
+	    // Pass to generic dispatcher if not found
+	    gr_block::dispatch_msg(which_port, msg);
+        }
+    }
 };
 
 /*!
diff --git a/gnuradio-core/src/lib/general/gr_feval.cc b/gnuradio-core/src/lib/general/gr_feval.cc
index ca5714a79..89f09984c 100644
--- a/gnuradio-core/src/lib/general/gr_feval.cc
+++ b/gnuradio-core/src/lib/general/gr_feval.cc
@@ -88,6 +88,22 @@ gr_feval::calleval(void)
   eval();
 }
 
+// ----------------------------------------------------------------
+
+gr_feval_p::~gr_feval_p(){}
+
+void
+gr_feval_p::eval(pmt::pmt_t x)
+{
+  // nop
+}
+
+void
+gr_feval_p::calleval(pmt::pmt_t x)
+{
+  eval(x);
+}
+
 /*
  * Trivial examples showing C++ (transparently) calling Python
  */
diff --git a/gnuradio-core/src/lib/general/gr_feval.h b/gnuradio-core/src/lib/general/gr_feval.h
index 1726a8a7f..a9bccfe51 100644
--- a/gnuradio-core/src/lib/general/gr_feval.h
+++ b/gnuradio-core/src/lib/general/gr_feval.h
@@ -24,6 +24,7 @@
 
 #include <gr_core_api.h>
 #include <gr_complex.h>
+#include <gruel/pmt.h>
 
 /*!
  * \brief base class for evaluating a function: double -> double
@@ -138,6 +139,34 @@ public:
 };
 
 /*!
+ * \brief base class for evaluating a function: pmt -> void
+ * \ingroup misc
+ *
+ * This class is designed to be subclassed in Python or C++
+ * and is callable from both places.  It uses SWIG's
+ * "director" feature to implement the magic.
+ * It's slow. Don't use it in a performance critical path.
+ *
+ * Override eval to define the behavior.
+ * Use calleval to invoke eval (this kludge is required to allow a
+ * python specific "shim" to be inserted.
+ */
+class GR_CORE_API gr_feval_p
+{
+protected:
+  /*!
+   * \brief override this to define the function
+   */
+  virtual void eval(pmt::pmt_t x);
+
+public:
+  gr_feval_p() {}
+  virtual ~gr_feval_p();
+
+  virtual void calleval(pmt::pmt_t x);	// invoke "eval"
+};
+
+/*!
  * \brief trivial examples / test cases showing C++ calling Python code
  */
 GR_CORE_API double     gr_feval_dd_example(gr_feval_dd *f, double x);
diff --git a/gnuradio-core/src/lib/general/gr_feval.i b/gnuradio-core/src/lib/general/gr_feval.i
index bc219a643..bcf4f1e64 100644
--- a/gnuradio-core/src/lib/general/gr_feval.i
+++ b/gnuradio-core/src/lib/general/gr_feval.i
@@ -45,23 +45,28 @@
 
 // Directors are only supported in Python, Java and C#
 #ifdef SWIGPYTHON
+%include "pmt_swig.i"
+using namespace pmt;
 
 // Enable SWIG directors for these classes
 %feature("director") gr_py_feval_dd;
 %feature("director") gr_py_feval_cc;
 %feature("director") gr_py_feval_ll;
 %feature("director") gr_py_feval;
+%feature("director") gr_py_feval_p;
 
 %feature("nodirector") gr_py_feval_dd::calleval;
 %feature("nodirector") gr_py_feval_cc::calleval;
 %feature("nodirector") gr_py_feval_ll::calleval;
 %feature("nodirector") gr_py_feval::calleval;
+%feature("nodirector") gr_py_feval_p::calleval;
 
 
 %rename(feval_dd) gr_py_feval_dd;
 %rename(feval_cc) gr_py_feval_cc;
 %rename(feval_ll) gr_py_feval_ll;
 %rename(feval)    gr_py_feval;
+%rename(feval_p)  gr_py_feval_p;
 
 //%exception {
 //  try { $action }
@@ -136,12 +141,26 @@ public:
   virtual void calleval();
 };
 
+%ignore gr_feval_p;
+class gr_feval_p
+{
+protected:
+  virtual void eval(pmt_t x);
+
+public:
+  gr_feval_p() {}
+  virtual ~gr_feval_p();
+
+  virtual void calleval(pmt_t x);
+};
+
 /*
  * These are the ones to derive from in Python.  They have the magic shim
  * that ensures that we're holding the Python GIL when we enter Python land...
  */
 
 %inline %{
+#include <gruel/pmt.h>
 
 class gr_py_feval_dd : public gr_feval_dd
 {
@@ -183,6 +202,16 @@ class gr_py_feval : public gr_feval
   }
 };
 
+class gr_py_feval_p : public gr_feval_p
+{
+ public:
+  void calleval(pmt::pmt_t x)
+  {
+    ensure_py_gil_state _lock;
+    eval(x);
+  }
+};
+
 %}
 
 
diff --git a/gnuradio-core/src/lib/general/gr_prefs.cc b/gnuradio-core/src/lib/general/gr_prefs.cc
index 20aead8e3..9705eed4f 100644
--- a/gnuradio-core/src/lib/general/gr_prefs.cc
+++ b/gnuradio-core/src/lib/general/gr_prefs.cc
@@ -25,6 +25,14 @@
 #endif
 
 #include <gr_prefs.h>
+#include <gr_sys_paths.h>
+#include <gr_constants.h>
+#include <algorithm>
+
+#include <boost/filesystem/operations.hpp>
+#include <boost/filesystem/path.hpp>
+#include <boost/filesystem/fstream.hpp>
+namespace fs = boost::filesystem;
 
 /*
  * Stub implementations
@@ -45,44 +53,177 @@ gr_prefs::set_singleton(gr_prefs *p)
   s_singleton = p;
 }
 
+gr_prefs::gr_prefs()
+{
+  _read_files();
+}
+
 gr_prefs::~gr_prefs()
 {
   // nop
 }
 
+std::vector<std::string>
+gr_prefs::_sys_prefs_filenames()
+{
+  std::vector<std::string> fnames;
+
+  fs::path dir = gr_prefsdir();
+  if(!fs::is_directory(dir))
+    return fnames;
+
+  fs::directory_iterator diritr(dir);
+  while(diritr != fs::directory_iterator()) {
+    fs::path p = *diritr++;
+    fnames.push_back(p.string());
+  }
+  std::sort(fnames.begin(), fnames.end());
+
+  // Find if there is a ~/.gnuradio/config file and add this to the
+  // beginning of the file list to override any preferences in the
+  // installed path config files.
+  fs::path homedir = fs::path(gr_appdata_path());
+  homedir = homedir/".gnuradio/config.conf";
+  if(fs::exists(homedir)) {
+    fnames.insert(fnames.begin(), homedir.string());
+  }    
+
+  return fnames;
+}
+
+void
+gr_prefs::_read_files()
+{
+  std::vector<std::string> filenames = _sys_prefs_filenames();
+  std::vector<std::string>::iterator sitr;
+  char tmp[1024];
+  for(sitr = filenames.begin(); sitr != filenames.end(); sitr++) {
+    fs::ifstream fin(*sitr);
+    while(!fin.eof()) {
+      fin.getline(tmp, 1024);
+      std::string t(tmp);
+      // ignore empty lines or lines of just comments
+      if((t.size() > 0) && (t[0] != '#')) {
+	// remove any comments in the line
+	size_t hash = t.find("#");
+
+	// Use hash marks at the end of each segment as a delimiter
+	d_configs += t.substr(0, hash) + '#';
+      }
+    }
+    fin.close();
+  }
+
+  // Remove all whitespace
+  d_configs.erase(std::remove_if(d_configs.begin(), d_configs.end(), ::isspace), d_configs.end());
+}
+
 bool
-gr_prefs::has_section(const std::string section)
+gr_prefs::has_section(const std::string &section)
 {
-  return false;
+  size_t t = d_configs.find("[" + section + "]#");
+  return t != std::string::npos;
 }
 
 bool
-gr_prefs::has_option(const std::string section, const std::string option)
+gr_prefs::has_option(const std::string &section, const std::string &option)
 {
-  return false;
+  if(has_section(section)) {
+    size_t sec = d_configs.find("[" + section + "]#");
+    size_t opt = d_configs.find("#" + option + "=", sec);
+    return opt != std::string::npos;
+  }
+  else {
+    return false;
+  }
 }
 
 const std::string
-gr_prefs::get_string(const std::string section, const std::string option, const std::string default_val)
+gr_prefs::get_string(const std::string &section, const std::string &option,
+		     const std::string &default_val)
 {
-  return default_val;
+  std::stringstream envname;
+  std::string secname=section, optname=option;
+
+  std::transform(section.begin(), section.end(), secname.begin(), ::toupper);
+  std::transform(option.begin(), option.end(), optname.begin(), ::toupper);
+  envname << "GR_CONF_" << secname << "_" << optname;
+
+  char *v = getenv(envname.str().c_str());
+  if(v) {
+    return std::string(v);
+  }
+
+  if(has_option(section, option)) {
+    std::string optname = "#" + option + "=";
+    size_t sec = d_configs.find("[" + section + "]#");
+    size_t opt = d_configs.find(optname, sec);
+
+    size_t start = opt + optname.size();
+    size_t end = d_configs.find("#", start);
+    size_t len = end - start;
+
+    return d_configs.substr(start, len);
+  }
+  else {
+    return default_val;
+  }
 }
 
 bool
-gr_prefs::get_bool(const std::string section, const std::string option, bool default_val)
+gr_prefs::get_bool(const std::string &section, const std::string &option, bool default_val)
 {
-  return default_val;
+  if(has_option(section, option)) {
+    std::string str = get_string(section, option, "");
+    if(str == "") {
+      return default_val;
+    }
+    std::transform(str.begin(), str.end(), str.begin(), ::tolower);
+    if((str == "true") || (str == "on") || (str == "1"))
+      return true;
+    else if((str == "false") || (str == "off") || (str == "0"))
+      return false;
+    else
+      return default_val;
+  }
+  else {
+    return default_val;
+  }
 }
 
 long
-gr_prefs::get_long(const std::string section, const std::string option, long default_val)
+gr_prefs::get_long(const std::string &section, const std::string &option, long default_val)
 {
-  return default_val;
+  if(has_option(section, option)) {
+    std::string str = get_string(section, option, "");
+    if(str == "") {
+      return default_val;
+    }
+    std::stringstream sstr(str);
+    long n;
+    sstr >> n;
+    return n;
+  }
+  else {
+    return default_val;
+  }
 }
 
 double
-gr_prefs::get_double(const std::string section, const std::string option, double default_val)
+gr_prefs::get_double(const std::string &section, const std::string &option, double default_val)
 {
-  return default_val;
+  if(has_option(section, option)) {
+    std::string str = get_string(section, option, "");
+    if(str == "") {
+      return default_val;
+    }
+    std::stringstream sstr(str);
+    double n;
+    sstr >> n;
+    return n;
+  }
+  else {
+    return default_val;
+  }
 }
 
diff --git a/gnuradio-core/src/lib/general/gr_prefs.h b/gnuradio-core/src/lib/general/gr_prefs.h
index b1c354bd3..dc2745d16 100644
--- a/gnuradio-core/src/lib/general/gr_prefs.h
+++ b/gnuradio-core/src/lib/general/gr_prefs.h
@@ -24,6 +24,7 @@
 
 #include <gr_core_api.h>
 #include <string>
+#include <gruel/thread.h>
 
 /*!
  * \brief Base class for representing user preferences a la windows INI files.
@@ -39,45 +40,54 @@ public:
   static gr_prefs *singleton();
   static void set_singleton(gr_prefs *p);
 
+  gr_prefs();
   virtual ~gr_prefs();
 
   /*!
    * \brief Does \p section exist?
    */
-  virtual bool has_section(const std::string section);
+  virtual bool has_section(const std::string &section);
 
   /*!
    * \brief Does \p option exist?
    */
-  virtual bool has_option(const std::string section, const std::string option);
+  virtual bool has_option(const std::string &section, const std::string &option);
 
   /*!
    * \brief If option exists return associated value; else default_val.
    */
-  virtual const std::string get_string(const std::string section,
-				       const std::string option,
-				       const std::string default_val);
+  virtual const std::string get_string(const std::string &section,
+				       const std::string &option,
+				       const std::string &default_val);
 
   /*!
    * \brief If option exists and value can be converted to bool, return it; else default_val.
    */
-  virtual bool get_bool(const std::string section,
-			const std::string option,
+  virtual bool get_bool(const std::string &section,
+			const std::string &option,
 			bool default_val);
 
   /*!
    * \brief If option exists and value can be converted to long, return it; else default_val.
    */
-  virtual long get_long(const std::string section,
-			const std::string option,
+  virtual long get_long(const std::string &section,
+			const std::string &option,
 			long default_val);
 
   /*!
    * \brief If option exists and value can be converted to double, return it; else default_val.
    */
-  virtual double get_double(const std::string section,
-			    const std::string option,
+  virtual double get_double(const std::string &section,
+			    const std::string &option,
 			    double default_val);
+
+ protected:
+  virtual std::vector<std::string> _sys_prefs_filenames();
+  virtual void _read_files();
+
+ private:
+  gruel::mutex d_mutex;
+  std::string d_configs;
 };
 
 
diff --git a/gnuradio-core/src/lib/general/gr_prefs.i b/gnuradio-core/src/lib/general/gr_prefs.i
index f44dcc944..b21d47f3b 100644
--- a/gnuradio-core/src/lib/general/gr_prefs.i
+++ b/gnuradio-core/src/lib/general/gr_prefs.i
@@ -20,9 +20,6 @@
  * Boston, MA 02110-1301, USA.
  */
 
-// Generate SWIG directors for gr_prefs.
-%feature("director") gr_prefs;
-
 class gr_prefs
 {
 public:
@@ -34,39 +31,39 @@ public:
   /*!
    * \brief Does \p section exist?
    */
-  virtual bool has_section(const std::string section);
+  virtual bool has_section(const std::string &section);
 
   /*!
    * \brief Does \p option exist?
    */
-  virtual bool has_option(const std::string section, const std::string option);
+  virtual bool has_option(const std::string &section, const std::string &option);
 
   /*!
    * \brief If option exists return associated value; else default_val.
    */
-  virtual const std::string get_string(const std::string section,
-				       const std::string option,
-				       const std::string default_val);
+  virtual const std::string get_string(const std::string &section,
+				       const std::string &option,
+				       const std::string &default_val);
 
   /*!
    * \brief If option exists and value can be converted to bool, return it; else default_val.
    */
-  virtual bool get_bool(const std::string section,
-			const std::string option,
+  virtual bool get_bool(const std::string &section,
+			const std::string &option,
 			bool default_val);
 
   /*!
    * \brief If option exists and value can be converted to long, return it; else default_val.
    */
-  virtual long get_long(const std::string section,
-			const std::string option,
+  virtual long get_long(const std::string &section,
+			const std::string &option,
 			long default_val);
 
   /*!
    * \brief If option exists and value can be converted to double, return it; else default_val.
    */
-  virtual double get_double(const std::string section,
-			    const std::string option,
+  virtual double get_double(const std::string &section,
+			    const std::string &option,
 			    double default_val);
 };
 
diff --git a/gnuradio-core/src/lib/general/gr_random_pdu.cc b/gnuradio-core/src/lib/general/gr_random_pdu.cc
new file mode 100644
index 000000000..6d8c13614
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_random_pdu.cc
@@ -0,0 +1,84 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2012 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.
+ */
+
+#ifdef HAVE_CONFIG_H
+#include "config.h"
+#endif
+
+#include <gr_random_pdu.h>
+#include <gr_io_signature.h>
+#include <cstdio>
+#include <errno.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <stdexcept>
+#include <string.h>
+#include <iostream>
+#include <vector>
+
+// public constructor that returns a shared_ptr
+
+gr_random_pdu_sptr
+gr_make_random_pdu (int items_min, int items_max)
+{
+  return gnuradio::get_initial_sptr(new gr_random_pdu(items_min, items_max));
+}
+
+gr_random_pdu::gr_random_pdu (int items_min, int items_max)
+  : gr_block("random_pdu",
+	     gr_make_io_signature(0, 0, 0),
+	     gr_make_io_signature(0, 0, 0)),
+    urange(items_min, items_max),
+    brange(0, 255),
+    rvar(rng, urange),
+    bvar(rng, brange)
+{
+    message_port_register_out(pmt::mp("pdus"));
+    message_port_register_in(pmt::mp("generate"));
+    set_msg_handler(pmt::mp("generate"), boost::bind(&gr_random_pdu::generate_pdu, this, _1));
+}
+
+bool gr_random_pdu::start(){
+    output_random();
+    return true;
+}
+
+void gr_random_pdu::output_random(){
+
+    // pick a random vector length
+    int len = rvar();
+
+    // fill it with random bytes
+    std::vector<unsigned char> vec;
+    for(int i=0; i<len; i++){
+        vec.push_back((unsigned char) bvar());
+        }
+
+    // send the vector
+    pmt::pmt_t vecpmt( pmt::pmt_make_blob( &vec[0], len ) );
+    pmt::pmt_t pdu( pmt::pmt_cons( pmt::PMT_NIL, vecpmt ) );
+    message_port_pub( pmt::mp("pdus"), pdu );
+    
+    std::cout << "sending new random vector of length " << len << "\n";
+}
+
diff --git a/gnuradio-core/src/lib/general/gr_random_pdu.h b/gnuradio-core/src/lib/general/gr_random_pdu.h
new file mode 100644
index 000000000..e6457d21b
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_random_pdu.h
@@ -0,0 +1,64 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2012 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.
+ */
+
+#ifndef INCLUDED_GR_RANDOM_PDU_H
+#define INCLUDED_GR_RANDOM_PDU_H
+
+#include <gr_core_api.h>
+#include <gr_block.h>
+#include <gr_message.h>
+#include <gr_msg_queue.h>
+
+#include <boost/random.hpp>
+#include <boost/generator_iterator.hpp>
+
+class gr_random_pdu;
+typedef boost::shared_ptr<gr_random_pdu> gr_random_pdu_sptr;
+
+GR_CORE_API gr_random_pdu_sptr gr_make_random_pdu (int mintime, int maxtime);
+
+/*!
+ * \brief Send message at defined interval
+ * \ingroup msg_blk
+ */
+class GR_CORE_API gr_random_pdu : public gr_block
+{
+ private:
+  friend GR_CORE_API gr_random_pdu_sptr
+  gr_make_random_pdu(int mintime, int maxtime);
+
+  void output_random();
+
+  boost::mt19937 rng;
+  boost::uniform_int<> urange;
+  boost::uniform_int<> brange;
+  boost::variate_generator< boost::mt19937, boost::uniform_int<> > rvar; // pdu length
+  boost::variate_generator< boost::mt19937, boost::uniform_int<> > bvar; // pdu contents
+
+ public:
+  gr_random_pdu (int, int);
+  bool start();
+  void generate_pdu(pmt::pmt_t msg){ output_random(); }
+  void generate_pdu(){ output_random(); }
+};
+
+#endif /* INCLUDED_GR_RANDOM_PDU_H */
diff --git a/gnuradio-core/src/lib/general/gr_random_pdu.i b/gnuradio-core/src/lib/general/gr_random_pdu.i
new file mode 100644
index 000000000..045a33060
--- /dev/null
+++ b/gnuradio-core/src/lib/general/gr_random_pdu.i
@@ -0,0 +1,30 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2005 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.
+ */
+
+GR_SWIG_BLOCK_MAGIC(gr,random_pdu);
+
+%{
+#include <gr_random_pdu.h>
+%}
+
+%include "gr_random_pdu.h"
+
diff --git a/gnuradio-core/src/lib/gengen/CMakeLists.txt b/gnuradio-core/src/lib/gengen/CMakeLists.txt
index b44a47075..db3103a26 100644
--- a/gnuradio-core/src/lib/gengen/CMakeLists.txt
+++ b/gnuradio-core/src/lib/gengen/CMakeLists.txt
@@ -85,6 +85,7 @@ expand_h_cc_i(gr_vector_source_X   b s i f c)
 expand_h_cc_i(gr_vector_insert_X   b)
 expand_h_cc_i(gr_vector_sink_X     b s i f c)
 expand_h_cc_i(gr_noise_source_X      s i f c)
+expand_h_cc_i(gr_fastnoise_source_X      s i f c)
 expand_h_cc_i(gr_sig_source_X        s i f c)
 expand_h_cc_i(gr_probe_signal_X    b s i f c)
 expand_h_cc_i(gr_probe_signal_vX    b s i f c)
diff --git a/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.cc.t b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.cc.t
new file mode 100644
index 000000000..7be7bdde8
--- /dev/null
+++ b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.cc.t
@@ -0,0 +1,116 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2004,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.
+ */
+
+// @WARNING@
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+#include <@NAME@.h>
+#include <gr_io_signature.h>
+#include <stdexcept>
+
+
+@NAME@_sptr
+gr_make_@BASE_NAME@ (gr_noise_type_t type, float ampl, long seed, long samples)
+{
+  return gnuradio::get_initial_sptr(new @NAME@ (type, ampl, seed, samples));
+}
+
+
+@NAME@::@NAME@ (gr_noise_type_t type, float ampl, long seed, long samples)
+  : gr_sync_block ("@BASE_NAME@",
+		   gr_make_io_signature (0, 0, 0),
+		   gr_make_io_signature (1, 1, sizeof (@TYPE@))),
+    d_type (type),
+    d_ampl (ampl),
+    d_rng (seed)
+{
+  d_samples.resize(samples);
+  generate();
+}
+
+void
+@NAME@::generate()
+{
+  int noutput_items = d_samples.size();
+  switch (d_type){
+#if @IS_COMPLEX@	// complex?
+
+  case GR_UNIFORM:
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = gr_complex (d_ampl * ((d_rng.ran1 () * 2.0) - 1.0),
+			   d_ampl * ((d_rng.ran1 () * 2.0) - 1.0));
+    break;
+
+  case GR_GAUSSIAN:
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = d_ampl * d_rng.rayleigh_complex ();
+    break;
+
+#else			// nope...
+
+  case GR_UNIFORM:
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = (@TYPE@)(d_ampl * ((d_rng.ran1 () * 2.0) - 1.0));
+    break;
+
+  case GR_GAUSSIAN:
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = (@TYPE@)(d_ampl * d_rng.gasdev ());
+    break;
+
+  case GR_LAPLACIAN:
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = (@TYPE@)(d_ampl * d_rng.laplacian ());
+    break;
+
+  case GR_IMPULSE:	// FIXME changeable impulse settings
+    for (int i = 0; i < noutput_items; i++)
+      d_samples[i] = (@TYPE@)(d_ampl * d_rng.impulse (9));
+    break;
+#endif
+
+  default:
+    throw std::runtime_error ("invalid type");
+  }
+
+}
+
+int
+@NAME@::work (int noutput_items,
+		   gr_vector_const_void_star &input_items,
+		   gr_vector_void_star &output_items)
+{
+  @TYPE@ *out = (@TYPE@ *) output_items[0];
+
+  for(int i=0; i<noutput_items; i++){
+#ifdef __USE_GNU
+    size_t idx = lrand48() % d_samples.size();
+#else
+    size_t idx = rand() % d_samples.size();
+#endif
+    out[i] = d_samples[idx];
+    }
+
+  return noutput_items;
+}
diff --git a/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.h.t b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.h.t
new file mode 100644
index 000000000..007e44975
--- /dev/null
+++ b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.h.t
@@ -0,0 +1,87 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2004 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.
+ */
+
+// @WARNING@
+
+#ifndef @GUARD_NAME@
+#define @GUARD_NAME@
+
+#include <gr_core_api.h>
+#include <gr_sync_block.h>
+#include <gr_noise_type.h>
+#include <gr_random.h>
+
+
+class @NAME@;
+typedef boost::shared_ptr<@NAME@> @NAME@_sptr;
+
+/*! \brief Make a noise source
+ * \param type  the random distribution to use (see gr_noise_type.h)
+ * \param ampl  a scaling factor for the output
+ * \param seed seed for random generators. Note that for uniform and
+ * Gaussian distributions, this should be a negative number.
+ * \param samples number of samples to pre-generate.
+ */
+GR_CORE_API @NAME@_sptr
+gr_make_@BASE_NAME@ (gr_noise_type_t type, float ampl, long seed = 0, long samples=1024*16);
+
+/*!
+ * \brief Random number source
+ * \ingroup source_blk
+ *
+ * \details
+ * Generate random values from different distributions.
+ * Currently, only Gaussian and uniform are enabled.
+ *
+ * \param type  the random distribution to use (see gr_noise_type.h)
+ * \param ampl  a scaling factor for the output
+ * \param seed seed for random generators. Note that for uniform and
+ * Gaussian distributions, this should be a negative number.
+ * \param samples number of samples to pre-generate.
+ */
+class GR_CORE_API @NAME@ : public gr_sync_block {
+  friend GR_CORE_API @NAME@_sptr
+
+  gr_make_@BASE_NAME@ (gr_noise_type_t type, float ampl, long seed, long samples);
+
+  gr_noise_type_t	d_type;
+  float			d_ampl;
+  gr_random		d_rng;
+  std::vector<@TYPE@> d_samples;
+
+  @NAME@ (gr_noise_type_t type, float ampl, long seed = 0, long samples=1024*16);
+
+ public:
+  void set_type (gr_noise_type_t type) { d_type = type; generate(); }
+  void set_amplitude (float ampl) { d_ampl = ampl; generate(); }
+  void generate();
+
+  gr_noise_type_t type () const { return d_type; }
+  float amplitude () const { return d_ampl; }
+
+  virtual int work (int noutput_items,
+			 gr_vector_const_void_star &input_items,
+			 gr_vector_void_star &output_items);
+};
+
+
+#endif
diff --git a/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.i.t b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.i.t
new file mode 100644
index 000000000..e1f7c775b
--- /dev/null
+++ b/gnuradio-core/src/lib/gengen/gr_fastnoise_source_X.i.t
@@ -0,0 +1,40 @@
+/* -*- c++ -*- */
+/*
+ * Copyright 2004 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.
+ */
+
+// @WARNING@
+
+GR_SWIG_BLOCK_MAGIC(gr,@BASE_NAME@);
+
+@NAME@_sptr
+gr_make_@BASE_NAME@ (gr_noise_type_t type, float ampl, long seed = 0, long samples = 1024*16);
+
+class @NAME@ : public gr_block {
+ private:
+  @NAME@ (gr_noise_type_t type, float ampl, long seed = 0, long samples = 1024*16);
+
+ public:
+  void set_type (gr_noise_type_t type) { d_type = type; }
+  void set_amplitude (float ampl) { d_ampl = ampl; }
+
+  gr_noise_type_t type () const { return d_type; }
+  float amplitude () const { return d_ampl; }
+};
diff --git a/gnuradio-core/src/lib/hier/gr_channel_model.cc b/gnuradio-core/src/lib/hier/gr_channel_model.cc
index bb01972d2..7da357809 100644
--- a/gnuradio-core/src/lib/hier/gr_channel_model.cc
+++ b/gnuradio-core/src/lib/hier/gr_channel_model.cc
@@ -59,7 +59,7 @@ gr_channel_model::gr_channel_model(double noise_voltage,
   d_multipath = gr_make_fir_filter_ccc(1, d_taps);
 
   d_noise_adder = gr_make_add_cc();
-  d_noise = gr_make_noise_source_c(GR_GAUSSIAN, noise_voltage, noise_seed);
+  d_noise = gr_make_fastnoise_source_c(GR_GAUSSIAN, noise_voltage, noise_seed, 1024*8);
   d_freq_offset = gr_make_sig_source_c(1, GR_SIN_WAVE, frequency_offset, 1.0, 0.0);
   d_mixer_offset = gr_make_multiply_cc();
 
diff --git a/gnuradio-core/src/lib/hier/gr_channel_model.h b/gnuradio-core/src/lib/hier/gr_channel_model.h
index f0b31b1a6..5796a6db2 100644
--- a/gnuradio-core/src/lib/hier/gr_channel_model.h
+++ b/gnuradio-core/src/lib/hier/gr_channel_model.h
@@ -25,7 +25,7 @@
 #include <gr_sig_source_c.h>
 #include <gr_fir_filter_ccc.h>
 #include <gr_add_cc.h>
-#include <gr_noise_source_c.h>
+#include <gr_fastnoise_source_c.h>
 #include <gr_multiply_cc.h>
 
 class gr_channel_model;
@@ -61,7 +61,7 @@ class GR_CORE_API gr_channel_model : public gr_hier_block2
   gr_sig_source_c_sptr d_freq_offset;
   gr_fir_filter_ccc_sptr d_multipath;
   gr_add_cc_sptr d_noise_adder;
-  gr_noise_source_c_sptr d_noise;
+  gr_fastnoise_source_c_sptr d_noise;
   gr_multiply_cc_sptr d_mixer_offset;
 
   std::vector<gr_complex> d_taps;
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 */
diff --git a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
index bd78c8fb4..da22a5f98 100644
--- a/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
+++ b/gnuradio-core/src/python/gnuradio/gr/CMakeLists.txt
@@ -29,6 +29,7 @@ GR_PYTHON_INSTALL(FILES
     gr_threading_24.py
     hier_block2.py
     prefs.py
+    tag_utils.py
     top_block.py
     pubsub.py
     DESTINATION ${GR_PYTHON_DIR}/gnuradio/gr
diff --git a/gnuradio-core/src/python/gnuradio/gr/__init__.py b/gnuradio-core/src/python/gnuradio/gr/__init__.py
index b274f8305..dab555052 100644
--- a/gnuradio-core/src/python/gnuradio/gr/__init__.py
+++ b/gnuradio-core/src/python/gnuradio/gr/__init__.py
@@ -29,6 +29,7 @@ from exceptions import *
 #from hier_block2 import *
 #from top_block import *
 from gateway import basic_block, sync_block, decim_block, interp_block
+from tag_utils import tag_to_python, tag_to_pmt
 
 import gras
 
@@ -55,7 +56,7 @@ serial_to_parallel = stream_to_vector
 parallel_to_serial = vector_to_stream
 
 # Force the preference database to be initialized
-from prefs import prefs
+prefs = gr_prefs.singleton
 
 #alias old gr_add_vXX and gr_multiply_vXX
 add_vcc = add_cc
diff --git a/gnuradio-core/src/python/gnuradio/gr/gateway.py b/gnuradio-core/src/python/gnuradio/gr/gateway.py
index ad99804d1..2f80c85dd 100644
--- a/gnuradio-core/src/python/gnuradio/gr/gateway.py
+++ b/gnuradio-core/src/python/gnuradio/gr/gateway.py
@@ -60,6 +60,24 @@ class gateway_handler(gr_core.feval_ll):
         return 0
 
 ########################################################################
+# Handler that does callbacks from C++
+########################################################################
+class msg_handler(gr_core.feval_p):
+
+    #dont put a constructor, it wont work
+
+    def init(self, callback):
+        self._callback = callback
+
+    def eval(self, arg):
+        try: self._callback(arg)
+        except Exception as ex:
+            print("handler caught exception: %s"%ex)
+            import traceback; traceback.print_exc()
+            raise ex
+        return 0
+
+########################################################################
 # The guts that make this into a gr block
 ########################################################################
 class gateway_block(object):
@@ -91,6 +109,9 @@ class gateway_block(object):
             self.__handler, name, gr_in_sig, gr_out_sig, work_type, factor)
         self.__message = self.__gateway.gr_block_message()
 
+        #dict to keep references to all message handlers
+        self.__msg_handlers = {}
+
         #register gr_block functions
         prefix = 'gr_block__'
         for attr in [x for x in dir(self.__gateway) if x.startswith(prefix)]:
@@ -174,6 +195,13 @@ class gateway_block(object):
     def start(self): return True
     def stop(self): return True
 
+    def set_msg_handler(self, which_port, handler_func):
+        handler = msg_handler()
+        handler.init(handler_func)
+        self.__gateway.set_msg_handler_feval(which_port, handler)
+        # Save handler object in class so it's not garbage collected
+        self.__msg_handlers[which_port] = handler
+
 ########################################################################
 # Wrappers for the user to inherit from
 ########################################################################
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
index c1110c10b..098aabb4a 100755
--- a/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_pdu.py
@@ -63,7 +63,6 @@ class test_pdu(gr_unittest.TestCase):
         #pmt.pmt_print(pi)
         #print "Stream to PDU output ports: "
         #pmt.pmt_print(po)
-        time.sleep(0.1)
 
         self.tb.connect(src, snk)
         self.tb.connect(src, snk2)
@@ -84,7 +83,7 @@ class test_pdu(gr_unittest.TestCase):
         src.to_basic_block()._post( port, msg )
 
         while(dbg.num_messages() < 1):
-            time.sleep(0.5)
+            time.sleep(0.1)
         self.tb.stop()
         self.tb.wait()
 
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py b/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py
new file mode 100644
index 000000000..06bb96947
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_python_message_passing.py
@@ -0,0 +1,123 @@
+#!/usr/bin/env python
+#
+# Copyright 2013 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.
+#
+
+from gnuradio import gr, gr_unittest
+try: import pmt
+except: from gruel import pmt
+import numpy
+import time
+
+# Simple block to generate messages
+class message_generator(gr.sync_block):
+    def __init__(self, msg_list, msg_interval):
+        gr.sync_block.__init__(
+            self,
+            name = "message generator",
+            in_sig = [numpy.float32],
+            out_sig = None
+        )
+        self.msg_list = msg_list
+        self.msg_interval = msg_interval
+        self.msg_ctr = 0
+        self.message_port_register_out(pmt.pmt_intern('out_port'))
+
+
+    def work(self, input_items, output_items):
+        inLen = len(input_items[0])
+        while self.msg_ctr < len(self.msg_list) and \
+                (self.msg_ctr * self.msg_interval) < \
+                (self.nitems_read(0) + inLen):
+            self.message_port_pub(pmt.pmt_intern('out_port'),
+                                  self.msg_list[self.msg_ctr])
+            self.msg_ctr += 1
+        return inLen
+
+# Simple block to consume messages
+class message_consumer(gr.sync_block):
+    def __init__(self):
+        gr.sync_block.__init__(
+            self,
+            name = "message consumer",
+            in_sig = None,
+            out_sig = None
+        )
+        self.msg_list = []
+        self.message_port_register_in(pmt.pmt_intern('in_port'))
+        self.set_msg_handler(pmt.pmt_intern('in_port'),
+                             self.handle_msg)
+
+    def handle_msg(self, msg):
+        # Create a new PMT from long value and put in list
+        self.msg_list.append(pmt.pmt_from_long(pmt.pmt_to_long(msg)))
+
+class test_python_message_passing(gr_unittest.TestCase):
+    
+    def setUp(self):
+        self.tb = gr.top_block()
+
+    def tearDown(self):
+        self.tb = None
+
+    def test_000(self):
+        num_msgs = 10
+        msg_interval = 1000
+        msg_list = []
+        for i in range(num_msgs):
+            msg_list.append(pmt.pmt_from_long(i))
+
+        # Create vector source with dummy data to trigger messages
+        src_data = []
+        for i in range(num_msgs*msg_interval):
+            src_data.append(float(i))
+        src = gr.vector_source_f(src_data, False)
+        msg_gen = message_generator(msg_list, msg_interval)
+        msg_cons = message_consumer()
+        
+        # Connect vector source to message gen
+        self.tb.connect(src, msg_gen)
+        
+        # Connect message generator to message consumer
+        self.tb.msg_connect(msg_gen, 'out_port', msg_cons, 'in_port')
+
+        # Verify that the messgae port query functions work
+        self.assertEqual(pmt.pmt_symbol_to_string(pmt.pmt_vector_ref(
+                    msg_gen.message_ports_out(), 0)), 'out_port')
+        self.assertEqual(pmt.pmt_symbol_to_string(pmt.pmt_vector_ref(
+                    msg_cons.message_ports_in(), 0)), 'in_port')
+        
+        # Run to verify message passing
+        self.tb.start()
+
+        # Wait for all messages to be sent
+        while msg_gen.msg_ctr < num_msgs:
+            time.sleep(0.5)
+        self.tb.stop()
+        self.tb.wait()               
+        
+        # Verify that the message consumer got all the messages
+        self.assertEqual(num_msgs, len(msg_cons.msg_list))
+        for i in range(num_msgs):
+            self.assertTrue(pmt.pmt_equal(msg_list[i], msg_cons.msg_list[i]))
+        
+if __name__ == '__main__':
+    gr_unittest.run(test_python_message_passing, 
+                    'test_python_message_passing.xml')
diff --git a/gnuradio-core/src/python/gnuradio/gr/qa_tag_utils.py b/gnuradio-core/src/python/gnuradio/gr/qa_tag_utils.py
new file mode 100755
index 000000000..ca1184979
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/qa_tag_utils.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python
+#
+# Copyright 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.
+#
+
+from gnuradio import gr, gr_unittest
+import tag_utils
+
+try:
+    import pmt_swig as pmt
+except ImportError:
+    import pmt
+
+class test_tag_utils (gr_unittest.TestCase):
+
+    def setUp (self):
+        self.tb = gr.top_block ()
+
+
+    def tearDown (self):
+        self.tb = None
+
+    def test_001(self):
+        t = gr.gr_tag_t()
+        t.offset = 10
+        t.key = pmt.pmt_string_to_symbol('key')
+        t.value = pmt.pmt_from_long(23)
+        t.srcid = pmt.pmt_from_bool(False)
+        pt = tag_utils.tag_to_python(t)
+        self.assertEqual(pt.key, 'key')
+        self.assertEqual(pt.value, 23)
+        self.assertEqual(pt.offset, 10)
+
+
+if __name__ == '__main__':
+    print 'hi'
+    gr_unittest.run(test_tag_utils, "test_tag_utils.xml")
+
diff --git a/gnuradio-core/src/python/gnuradio/gr/tag_utils.py b/gnuradio-core/src/python/gnuradio/gr/tag_utils.py
new file mode 100644
index 000000000..923718fc9
--- /dev/null
+++ b/gnuradio-core/src/python/gnuradio/gr/tag_utils.py
@@ -0,0 +1,54 @@
+#
+# Copyright 2003-2012 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.
+#
+""" Conversion tools between stream tags and Python objects """
+
+try: import pmt
+except: from gruel import pmt
+
+from gnuradio_core import gr_tag_t
+
+class PythonTag(object):
+    " Python container for tags "
+    def __init__(self):
+        self.offset = None
+        self.key    = None
+        self.value  = None
+        self.srcid  = None
+
+def tag_to_python(tag):
+    """ Convert a stream tag to a Python-readable object """
+    newtag = PythonTag()
+    newtag.offset = tag.offset
+    newtag.key = pmt.to_python(tag.key)
+    newtag.value = pmt.to_python(tag.value)
+    newtag.srcid = pmt.to_python(tag.srcid)
+    return newtag
+
+def tag_to_pmt(tag):
+    """ Convert a Python-readable object to a stream tag """
+    newtag = gr_tag_t()
+    newtag.offset = tag.offset
+    newtag.key = pmt.to_python(tag.key)
+    newtag.value = pmt.from_python(tag.value)
+    newtag.srcid = pmt.from_python(tag.srcid)
+    return newtag
+
+