Merge branch 'max_nout_each'

8 files changed, 357 insertions, 2 deletions

diff --git a/docs/doxygen/other/main_page.dox b/docs/doxygen/other/main_page.dox
index 8f69a9737..c90ad883f 100644
--- a/docs/doxygen/other/main_page.dox
+++ b/docs/doxygen/other/main_page.dox
@@ -39,6 +39,230 @@ More details on packages in GNU Radio:
 \li \ref page_vocoder
 \li \ref page_pfb
 
+
+\section flowgraph Operating a Flowgraph
+
+The basic data structure in GNU Radio is the flowgraph, which
+represents the connections of the blocks through which a continuous
+stream of samples flows. The concept of a flowgraph is an acyclic
+directional graph with one or more source blocks (to insert samples
+into the flowgraph), one or more sink blocks (to terminate or export
+samples from the flowgraph), and any signal processing blocks in
+between.
+
+A program must at least create a GNU Radio 'top_block', which
+represents the top-most structure of the flowgraph. The top blocks
+provide the overall control and hold methods such as 'start,' 'stop,'
+and 'wait.'
+
+The general construction of a GNU Radio application is to create a
+gr_top_block, instantiate the blocks, connect the blocks together, and
+then start the gr_top_block. The following program shows how this is
+done. A single source and sink are used with a FIR filter between
+them.
+
+\code
+    from gnuradio import gr, filter
+    
+    class my_topblock(gr.top_block):
+        def __init__(self):
+            gr.top_block.__init__(self)
+    
+            amp = 1
+            taps = filter.firdes.low_pass(1, 1, 0.1, 0.01)
+            
+            self.src = gr.noise_source_c(gr.GR_GAUSSIAN, amp)
+            self.flt = filter.fir_filter_ccf(1, taps)
+            self.snk = gr.null_sink(gr.sizeof_gr_complex)
+    
+            self.connect(self.src, self.flt, self.snk)
+    
+    if __name__ == "__main__":
+        tb = my_topblock()
+        tb.start()
+        tb.wait()
+\endcode
+
+The 'tb.start()' starts the data flowing through the flowgraph while
+the 'tb.wait()' is the equivalent of a thread's 'join' operation and
+blocks until the gr_top_block is done.
+
+An alternative to using the 'start' and 'wait' methods, a 'run' method is
+also provided for convenience that is a blocking start call;
+equivalent to the above 'start' followed by a 'wait.'
+
+
+\subsection latency Latency and Throughput
+
+By default, GNU Radio runs a scheduler that attempts to optimize
+throughput. Using a dynamic scheduler, blocks in a flowgraph pass
+chunks of items from sources to sinks. The sizes of these chunks will
+vary depending on the speed of processing. For each block, the number
+of items is can process is dependent on how much space it has in its
+output buffer(s) and how many items are available on the input
+buffer(s).
+
+The consequence of this is that often a block may be called with a very
+large number of items to process (several thousand). In terms of
+speed, this is efficient since now the majority of the processing time
+is taken up with processing samples. Smaller chunks mean more calls
+into the scheduler to retrieve more data. The downside to this is that
+it can lead to large latency while a block is processing a large chunk
+of data.
+
+To combat this problem, the gr_top_block can be passed a limit on the
+number of output items a block will ever receive. A block may get less
+than this number, but never more, and so it serves as an upper limit
+to the latency any block will exhibit. By limiting the number of items
+per call to a block, though, we increase the overhead of the
+scheduler, and so reduce the overall efficiency of the application.
+
+To set the maximum number of output items, we pass a value into the
+'start' or 'run' method of the gr_top_block:
+
+\code
+     tb.start(1000)
+     tb.wait()
+or
+     tb.run(1000)
+\endcode
+
+Using this method, we place a global restriction on the size of items
+to all blocks. Each block, though, has the ability to overwrite this
+with its own limit. Using the 'set_max_noutput_items(m)' method for an
+individual block will overwrite the global setting. For example, in
+the following code, the global setting is 1000 items max, except for
+the FIR filter, which can receive up to 2000 items.
+
+\code
+     tb.flt.set_max_noutput_items(2000)
+     tb.run(1000)
+\endcode
+
+
+\section reconfigure Reconfiguring Flowgraphs
+
+It is possible to reconfigure the flowgraph at runtime. The
+reconfiguration is meant for changes in the flowgraph structure, not
+individual parameter settings of the blocks. For example, changing the
+constant in a gr_add_const_cc block can be done while the flowgraph is
+running using the 'set_k(k)' method.
+
+Reconfiguration is done by locking the flowgraph, which stops it from
+running and processing data, performing the reconfiguration, and then
+restarting the graph by unlocking it.
+
+The following example code shows a graph that first adds two
+gr_noise_source_c blocks and then replaces the gr_add_cc block with a
+gr_sub_cc block to then subtract the sources.
+
+\code
+from gnuradio import gr
+import time
+
+class mytb(gr.top_block):
+    def __init__(self):
+        gr.top_block.__init__(self)
+
+        self.src0 = gr.noise_source_c(gr.GR_GAUSSIAN, 1)
+        self.src1 = gr.noise_source_c(gr.GR_GAUSSIAN, 1)
+        self.add  = gr.add_cc()
+        self.sub  = gr.sub_cc()
+        self.head = gr.head(gr.sizeof_gr_complex, 1000000)
+        self.snk  = gr.file_sink(gr.sizeof_gr_complex, "output.32fc")
+
+        self.connect(self.src0, (self.add,0))
+        self.connect(self.src1, (self.add,1))
+        self.connect(self.add, self.head)
+        self.connect(self.head, self.snk)
+
+def main():
+    tb = mytb()
+    tb.start()
+    time.sleep(0.01)
+
+    # Stop flowgraph and disconnect the add block
+    tb.lock()
+    tb.disconnect(tb.add, tb.head)
+    tb.disconnect(tb.src0, (tb.add,0))
+    tb.disconnect(tb.src1, (tb.add,1))
+
+    # Connect the sub block and restart
+    tb.connect(tb.sub, tb.head)
+    tb.connect(tb.src0, (tb.sub,0))
+    tb.connect(tb.src1, (tb.sub,1))
+    tb.unlock()
+
+    tb.wait()
+
+if __name__ == "__main__":
+    main()
+\endcode
+
+During reconfiguration, the maximum noutput_items value can be changed
+either globally using the 'set_max_noutput_items(m)' on the gr_top_block
+object or locally using the 'set_max_noutput_items(m)' on any given
+block object.
+
+A block also has a 'unset_max_noutput_items()' method that unsets the
+local max noutput_items value so that block reverts back to using the
+global value.
+
+The following example expands the previous example but sets and resets
+the max noutput_items both locally and globally.
+
+\code
+from gnuradio import gr
+import time
+
+class mytb(gr.top_block):
+    def __init__(self):
+        gr.top_block.__init__(self)
+
+        self.src0 = gr.noise_source_c(gr.GR_GAUSSIAN, 1)
+        self.src1 = gr.noise_source_c(gr.GR_GAUSSIAN, 1)
+        self.add  = gr.add_cc()
+        self.sub  = gr.sub_cc()
+        self.head = gr.head(gr.sizeof_gr_complex, 1000000)
+        self.snk  = gr.file_sink(gr.sizeof_gr_complex, "output.32fc")
+
+        self.connect(self.src0, (self.add,0))
+        self.connect(self.src1, (self.add,1))
+        self.connect(self.add, self.head)
+        self.connect(self.head, self.snk)
+
+def main():
+    # Start the gr_top_block after setting some max noutput_items.
+    tb = mytb()
+    tb.src1.set_max_noutput_items(2000)
+    tb.start(100)
+    time.sleep(0.01)
+    
+    # Stop flowgraph and disconnect the add block
+    tb.lock()
+
+    tb.disconnect(tb.add, tb.head)
+    tb.disconnect(tb.src0, (tb.add,0))
+    tb.disconnect(tb.src1, (tb.add,1))
+
+    # Connect the sub block
+    tb.connect(tb.sub, tb.head)
+    tb.connect(tb.src0, (tb.sub,0))
+    tb.connect(tb.src1, (tb.sub,1))
+
+    # Set new max_noutput_items for the gr_top_block
+    # and unset the local value for src1
+    tb.set_max_noutput_items(1000)
+    tb.src1.unset_max_noutput_items()
+    tb.unlock()
+
+    tb.wait()
+
+if __name__ == "__main__":
+    main()
+\endcode
+
+
 \section volk_main Using Volk in GNU Radio
 
 The \ref volk_guide page provides an overview of how to incorporate
@@ -48,4 +272,5 @@ Many blocks have already been converted to use Volk in their calls, so
 they can also serve as examples. See the gr_complex_to_xxx.h file for
 examples of various blocks that make use of Volk.
 
+
 */
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.i b/gnuradio-core/src/lib/runtime/gr_block.i
index 4cc260bfe..e9341e8cb 100644
--- a/gnuradio-core/src/lib/runtime/gr_block.i
+++ b/gnuradio-core/src/lib/runtime/gr_block.i
@@ -52,6 +52,12 @@ class gr_block : public gr_basic_block {
   uint64_t nitems_read(unsigned int which_input);
   uint64_t nitems_written(unsigned int which_output);
 
+  // Methods to manage the block's max_noutput_items size.
+  int max_noutput_items();
+  void set_max_noutput_items(int m);
+  void unset_max_noutput_items();
+  bool is_set_max_noutput_items();
+
   // 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_executor.cc b/gnuradio-core/src/lib/runtime/gr_block_executor.cc
index c08534475..375b58f56 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 */