6 files changed, 126 insertions, 2 deletions
diff --git a/gnuradio-core/src/lib/runtime/gr_block.cc b/gnuradio-core/src/lib/runtime/gr_block.cc
index 9a5255a93..7e01c0ba8 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block.cc
@@ -40,6 +40,8 @@ gr_block::gr_block (const std::string &name,
     d_relative_rate (1.0),
     d_history(1),
     d_fixed_rate(false),
+    d_max_noutput_items_set(false),
+    d_max_noutput_items(0),
     d_tag_propagation_policy(TPP_ALL_TO_ALL)
 {
 }
@@ -208,6 +210,35 @@ gr_block::set_tag_propagation_policy(tag_propagation_policy_t p)
   d_tag_propagation_policy = p;
 }
 
+
+int
+gr_block::max_noutput_items()
+{
+  return d_max_noutput_items;
+}
+
+void
+gr_block::set_max_noutput_items(int m)
+{
+  if(m <= 0)
+    throw std::runtime_error("gr_block::set_max_noutput_items: value for max_noutput_items must be greater than 0.\n");
+
+  d_max_noutput_items = m;
+  d_max_noutput_items_set = true;
+}
+
+void
+gr_block::unset_max_noutput_items()
+{
+  d_max_noutput_items_set = false;
+}
+
+bool
+gr_block::is_set_max_noutput_items()
+{
+  return d_max_noutput_items_set;
+}
+
 std::ostream&
 operator << (std::ostream& os, const gr_block *m)
 {
diff --git a/gnuradio-core/src/lib/runtime/gr_block.h b/gnuradio-core/src/lib/runtime/gr_block.h
index 71ac8eee6..c89138bb3 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.h
+++ b/gnuradio-core/src/lib/runtime/gr_block.h
@@ -251,6 +251,43 @@ class GR_CORE_API gr_block : public gr_basic_block {
    */
   void set_tag_propagation_policy(tag_propagation_policy_t p);
 
+  /*!
+   * \brief Return the maximum number of output items this block will
+   * handle during a call to work.
+   */
+  int max_noutput_items();
+
+  /*!
+   * \brief Set the maximum number of ouput items htis block will
+   * handle during a call to work.
+   *
+   * \param m the maximum noutput_items this block will handle.
+   */
+  void set_max_noutput_items(int m);
+
+  /*!
+   * \brief Clear the switch for using the max_noutput_items value of this block.
+   *
+   * When is_set_max_noutput_items() returns 'true', the scheduler
+   * will use the value returned by max_noutput_items() to limit the
+   * size of the number of items possible for this block's work
+   * function. If is_set_max_notput_items() returns 'false', then the
+   * scheduler ignores the internal value and uses the value set
+   * globally in the top_block.
+   *
+   * Use this value to clear the 'is_set' flag so the scheduler will
+   * ignore this. Use the set_max_noutput_items(m) call to both set a
+   * new value for max_noutput_items and to reenable its use in the
+   * scheduler.
+   */
+  void unset_max_noutput_items();
+
+  /*!
+   * \brief Ask the block if the flag is or is not set to use the
+   * internal value of max_noutput_items during a call to work.
+   */
+  bool is_set_max_noutput_items();
+
   // ----------------------------------------------------------------------------
 
  private:
@@ -263,6 +300,8 @@ class GR_CORE_API gr_block : public gr_basic_block {
   gr_block_detail_sptr	d_detail;		// implementation details
   unsigned              d_history;
   bool                  d_fixed_rate;
+  bool                  d_max_noutput_items_set;     // if d_max_noutput_items is valid
+  int                   d_max_noutput_items;         // value of max_noutput_items for this block
   tag_propagation_policy_t d_tag_propagation_policy; // policy for moving tags downstream
 
  protected:
diff --git a/gnuradio-core/src/lib/runtime/gr_block_executor.cc b/gnuradio-core/src/lib/runtime/gr_block_executor.cc
index 6fea14613..5b24e9f6e 100644
--- a/gnuradio-core/src/lib/runtime/gr_block_executor.cc
+++ b/gnuradio-core/src/lib/runtime/gr_block_executor.cc
@@ -206,7 +206,7 @@ gr_block_executor::run_one_iteration()
 
     // determine the minimum available output space
     noutput_items = min_available_space (d, m->output_multiple ());
-    noutput_items = std::min(noutput_items, d_max_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
       goto were_done;
@@ -251,7 +251,7 @@ gr_block_executor::run_one_iteration()
     // take a swag at how much output we can sink
     noutput_items = (int) (max_items_avail * m->relative_rate ());
     noutput_items = round_down (noutput_items, m->output_multiple ());
-    noutput_items = std::min(noutput_items, d_max_noutput_items);
+    noutput_items = std::min(noutput_items, max_noutput_items);
     LOG(*d_log << "  max_items_avail = " << max_items_avail << std::endl);
     LOG(*d_log << "  noutput_items = " << noutput_items << std::endl);
 
diff --git a/gnuradio-core/src/lib/runtime/gr_scheduler_tpb.cc b/gnuradio-core/src/lib/runtime/gr_scheduler_tpb.cc
index 131ddd19c..2824eb1b3 100644
--- a/gnuradio-core/src/lib/runtime/gr_scheduler_tpb.cc
+++ b/gnuradio-core/src/lib/runtime/gr_scheduler_tpb.cc
@@ -74,6 +74,11 @@ gr_scheduler_tpb::gr_scheduler_tpb(gr_flat_flowgraph_sptr ffg, int max_noutput_i
   for (size_t i = 0; i < blocks.size(); i++){
     std::stringstream name;
     name << "thread-per-block[" << i << "]: " << blocks[i];
+
+    // If set, use internal value instead of global value
+    if(blocks[i]->is_set_max_noutput_items())
+      max_noutput_items = blocks[i]->max_noutput_items();
+    
     d_threads.create_thread(
 	    gruel::thread_body_wrapper<tpb_container>(tpb_container(blocks[i], max_noutput_items),
 						      name.str()));
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 cc7b7c720..fe0883eba 100644
--- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc
+++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.cc
@@ -119,3 +119,48 @@ void qa_gr_top_block::t4_reconfigure()
   // Wait for flowgraph to end on its own
   tb->wait();
 }
+
+
+void qa_gr_top_block::t5_max_noutputs()
+{
+  if (VERBOSE) std::cout << "qa_gr_top_block::t5()\n";
+
+  gr_top_block_sptr tb = gr_make_top_block("top");
+
+  gr_block_sptr src = gr_make_null_source(sizeof(int));
+  gr_block_sptr head = gr_make_head(sizeof(int), 100000);
+  gr_block_sptr dst = gr_make_null_sink(sizeof(int));
+
+  // Start infinite flowgraph
+  tb->connect(src, 0, head, 0);
+  tb->connect(head, 0, dst, 0);
+  tb->start(100);
+  tb->wait();
+}
+
+void qa_gr_top_block::t6_reconfig_max_noutputs()
+{
+  if (VERBOSE) std::cout << "qa_gr_top_block::t5()\n";
+
+  gr_top_block_sptr tb = gr_make_top_block("top");
+
+  gr_block_sptr src = gr_make_null_source(sizeof(int));
+  gr_block_sptr head = gr_make_head(sizeof(int), 100000);
+  gr_block_sptr dst = gr_make_null_sink(sizeof(int));
+
+  // Start infinite flowgraph
+  tb->connect(src, 0, dst, 0);
+  tb->start(100);
+
+  // Reconfigure with gr_head in the middle
+  tb->lock();
+  tb->disconnect(src, 0, dst, 0);
+  tb->connect(src, 0, head, 0);
+  tb->connect(head, 0, dst, 0);
+  tb->set_max_noutput_items(1000);
+  head->set_max_noutput_items(500);
+  tb->unlock();
+
+  // Wait for flowgraph to end on its own
+  tb->wait();
+}
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 b223633e5..7299ca330 100644
--- a/gnuradio-core/src/lib/runtime/qa_gr_top_block.h
+++ b/gnuradio-core/src/lib/runtime/qa_gr_top_block.h
@@ -36,6 +36,8 @@ class qa_gr_top_block : public CppUnit::TestCase
   CPPUNIT_TEST(t2_start_stop_wait);
   CPPUNIT_TEST(t3_lock_unlock);
   CPPUNIT_TEST(t4_reconfigure);  // triggers 'join never returns' bug
+  CPPUNIT_TEST(t5_max_noutputs);
+  CPPUNIT_TEST(t6_reconfig_max_noutputs);
 
   CPPUNIT_TEST_SUITE_END();
 
@@ -46,6 +48,8 @@ private:
   void t2_start_stop_wait();
   void t3_lock_unlock();
   void t4_reconfigure();
+  void t5_max_noutputs();
+  void t6_reconfig_max_noutputs();
 };
 
 #endif /* INCLUDED_QA_GR_TOP_BLOCK_H */