handle_update_inputs also performs resize, cleanup input fixed rate logic

3 files changed, 18 insertions, 1 deletions

diff --git a/TODO.txt b/TODO.txt
index 1564b59..59c8cd4 100644
--- a/TODO.txt
+++ b/TODO.txt
@@ -37,3 +37,8 @@ and not directly from block::<method>
 
 * want per-port token_pool for inputs and outputs
 ** if ports get removed, this will release subscribers
+
+* forecast + input buffer accumulate
+
+* token check for top block stop
+** fixes qa UDP so we dont wait for inert message
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 1570064..fac70c0 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -114,10 +114,17 @@ void BlockActor::handle_task(void)
         this->input_items[i].size() = items;
         this->work_input_items[i] = mem;
         this->work_ninput_items[i] = items;
+
         if (this->enable_fixed_rate)
         {
-            items = std::max<int>(0, int(items) - int(this->input_configs[i].lookahead_items));
+            if (items <= this->input_configs[i].lookahead_items)
+            {
+                if (this->inputs_done[i]) this->mark_done();
+                return;
+            }
+            items -= this->input_configs[i].lookahead_items;
         }
+
         num_input_items = std::min(num_input_items, items);
         this->consume_items[i] = 0;
         this->consume_called[i] = false;
@@ -181,6 +188,7 @@ void BlockActor::handle_task(void)
     //------------------------------------------------------------------
     //-- forecast
     //------------------------------------------------------------------
+    VAR(work_noutput_items);
     if (this->forecast_enable)
     {
         forecast_again_you_jerk:
@@ -207,6 +215,8 @@ void BlockActor::handle_task(void)
     //------------------------------------------------------------------
     //-- the work
     //------------------------------------------------------------------
+    VAR(work_noutput_items);
+    if (num_inputs) VAR(work_ninput_items[0]);
     this->work_ret = -1;
     if (this->interruptible_thread)
     {
diff --git a/lib/topology_handler.cpp b/lib/topology_handler.cpp
index ae9f040..eba9bca 100644
--- a/lib/topology_handler.cpp
+++ b/lib/topology_handler.cpp
@@ -106,7 +106,9 @@ void BlockActor::handle_update_inputs(
     const UpdateInputsMessage &,
     const Theron::Address
 ){
+    MESSAGE_TRACER();
     const size_t num_inputs = this->get_num_inputs();
+    this->input_queues.resize(num_inputs);
 
     //impose input reserve requirements based on relative rate and output multiple
     resize_fill_grow(this->input_reserve_items, num_inputs, 1);