reimplement the output multiple items/work output items calculations

5 files changed, 42 insertions, 19 deletions

diff --git a/lib/block.cpp b/lib/block.cpp
index fefec9f..f236ca4 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -56,6 +56,7 @@ Block::Block(const std::string &name):
     this->set_tag_propagation_policy(TPP_ALL_TO_ALL);
     this->set_interruptible_work(false);
     this->set_buffer_affinity(-1);
+    (*this)->block->output_multiple_items = 1;
 }
 
 template <typename V, typename T>
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 39b0327..c1046c3 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -142,30 +142,53 @@ void BlockActor::handle_task(void)
         const SBuffer &buff = this->output_queues.front(i);
         void *mem = buff.get(buff.length);
         const size_t bytes = buff.get_actual_length() - buff.length - buff.offset;
-        const size_t items = bytes/this->output_items_sizes[i];
+        size_t items = bytes/this->output_items_sizes[i];
 
         this->work_io_ptr_mask |= ptrdiff_t(mem);
         this->output_items[i].get() = mem;
         this->output_items[i].size() = items;
         this->work_output_items[i] = mem;
+
+        items /= this->output_multiple_items;
+        items *= this->output_multiple_items;
         num_output_items = std::min(num_output_items, items);
         this->produce_items[i] = 0;
     }
 
     //------------------------------------------------------------------
+    //-- calculate the work_noutput_items given:
+    //-- min of num_input_items
+    //-- min of num_output_items
+    //-- relative rate and output multiple items
+    //------------------------------------------------------------------
+    work_noutput_items = num_output_items;
+    if (num_inputs and (this->enable_fixed_rate or not num_outputs))
+    {
+        size_t calc_output_items = size_t(num_input_items*this->relative_rate);
+        calc_output_items += this->output_multiple_items-1;
+        calc_output_items /= this->output_multiple_items;
+        calc_output_items *= this->output_multiple_items;
+        if (calc_output_items and calc_output_items < work_noutput_items)
+            work_noutput_items = calc_output_items;
+    }
+
+    //------------------------------------------------------------------
     //-- forecast
     //------------------------------------------------------------------
     if (this->forecast_enable)
     {
         forecast_again_you_jerk:
-        fcast_ninput_items = work_ninput_items;
-        block_ptr->forecast(num_output_items, fcast_ninput_items);
+        fcast_ninput_items = work_ninput_items; //init for NOP case
+        block_ptr->forecast(work_noutput_items, fcast_ninput_items);
         for (size_t i = 0; i < num_inputs; i++)
         {
             if (fcast_ninput_items[i] <= work_ninput_items[i]) continue;
 
-            num_output_items = num_output_items/2; //backoff regime
-            if (num_output_items) goto forecast_again_you_jerk;
+            work_noutput_items = work_noutput_items/2; //backoff regime
+            work_noutput_items += this->output_multiple_items-1;
+            work_noutput_items /= this->output_multiple_items;
+            work_noutput_items *= this->output_multiple_items;
+            if (work_noutput_items) goto forecast_again_you_jerk;
 
             //handle the case of forecast failing
             this->mark_done();
@@ -176,9 +199,6 @@ void BlockActor::handle_task(void)
     //------------------------------------------------------------------
     //-- the work
     //------------------------------------------------------------------
-    work_noutput_items = num_output_items;
-    if (this->enable_fixed_rate) work_noutput_items = std::min(
-        work_noutput_items, myulround((num_input_items)*this->relative_rate));
     this->work_ret = -1;
     if (this->interruptible_thread)
     {
diff --git a/lib/gr_block.cpp b/lib/gr_block.cpp
index dea455c..834bbd5 100644
--- a/lib/gr_block.cpp
+++ b/lib/gr_block.cpp
@@ -46,6 +46,14 @@ int gr_block::work(
     );
 }
 
+void gr_block::forecast(int noutput_items, std::vector<int> &ninputs_req)
+{
+    for (size_t i = 0; i < ninputs_req.size(); i++)
+    {
+        ninputs_req[i] = fixed_rate_noutput_to_ninput(noutput_items);
+    }
+}
+
 int gr_block::general_work(
     int noutput_items,
     gr_vector_int &ninput_items,
@@ -72,8 +80,7 @@ bool gr_block::is_unaligned(void)
 
 size_t gr_block::fixed_rate_noutput_to_ninput(const size_t noutput_items)
 {
-    return (*this)->block->input_configs[0].lookahead_items +
-            size_t((noutput_items/this->relative_rate()));
+    return size_t((noutput_items/this->relative_rate()));
 }
 
 size_t gr_block::interpolation(void) const
@@ -113,11 +120,12 @@ void gr_block::set_history(unsigned history)
 
 unsigned gr_block::output_multiple(void) const
 {
-    return this->output_config().reserve_items+1;
+    return (*this)->block->output_multiple_items;
 }
 
 void gr_block::set_output_multiple(unsigned multiple)
 {
+    (*this)->block->output_multiple_items = multiple;
     gnuradio::OutputPortConfig config = this->output_config();
     config.reserve_items = multiple;
     this->set_output_config(config);
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index f11c569..5dd7421 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -131,6 +131,7 @@ struct BlockActor : Apology::Worker
     std::vector<InputPortConfig> input_configs;
     std::vector<OutputPortConfig> output_configs;
     std::vector<size_t> input_reserve_items;
+    size_t output_multiple_items;
 
     //keeps track of production
     std::vector<uint64_t> items_consumed;
diff --git a/lib/topology_handler.cpp b/lib/topology_handler.cpp
index 7e0c710..cc57ed6 100644
--- a/lib/topology_handler.cpp
+++ b/lib/topology_handler.cpp
@@ -107,7 +107,6 @@ void BlockActor::handle_update_inputs(
     const Theron::Address
 ){
     const size_t num_inputs = this->get_num_inputs();
-    const size_t num_outputs = this->get_num_outputs();
 
     //impose input reserve requirements based on relative rate and output multiple
     resize_fill_grow(this->input_reserve_items, num_inputs, 1);
@@ -115,13 +114,7 @@ void BlockActor::handle_update_inputs(
     for (size_t i = 0; i < num_inputs; i++)
     {
         input_lookahead_items[i] = this->input_configs[i].lookahead_items;
-
-        //TODO, this is a little cheap, we only look at output multiple [0]
-        const size_t multiple = (num_outputs)?this->output_configs.front().reserve_items:1;
-        if (this->enable_fixed_rate)
-        {
-            this->input_reserve_items[i] = size_t(std::ceil(multiple/this->relative_rate));
-        }
+        this->input_reserve_items[i] = size_t(std::ceil(this->output_multiple_items/this->relative_rate));
         if (this->input_reserve_items[i] == 0) this->input_reserve_items[i] = 1;
     }