input buffer history work and fixes

added consume flag to know if user called consume

fixes for mini history buff and flag when in use

6 files changed, 24 insertions, 6 deletions

diff --git a/lib/block.cpp b/lib/block.cpp
index 39939cb..7ffcbf3 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -92,6 +92,7 @@ void Block::set_output_multiple(const size_t multiple, const size_t which_output
 void Block::consume(const size_t which_input, const size_t how_many_items)
 {
     (*this)->consume_items[which_input] = how_many_items;
+    (*this)->consume_called[which_input] = true;
 }
 
 void Block::consume_each(const size_t how_many_items)
@@ -99,6 +100,7 @@ void Block::consume_each(const size_t how_many_items)
     for (size_t i = 0; i < (*this)->consume_items.size(); i++)
     {
         (*this)->consume_items[i] = how_many_items;
+        (*this)->consume_called[i] = true;
     }
 }
 
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index 2cb167f..243d158 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -166,6 +166,7 @@ void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface)
     this->input_items.resize(num_inputs);
     this->output_items.resize(num_outputs);
     this->consume_items.resize(num_inputs, 0);
+    this->consume_called.resize(num_inputs, false);
     this->produce_items.resize(num_outputs, 0);
     this->input_queues.resize(num_inputs);
     this->output_queues.resize(num_outputs);
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 114c073..c3ab0d4 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -124,6 +124,7 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
         this->work_input_items[i] = mem;
         this->work_ninput_items[i] = items;
         num_input_items = std::min(num_input_items, items);
+        this->consume_called[i] = false;
 
         //inline dealings, how and when input buffers can be inlined into output buffers
         //TODO, check that the buff.get_affinity() matches this block or we dont inline
@@ -223,7 +224,7 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
     for (size_t i = 0; i < num_inputs; i++)
     {
         ASSERT(enable_fixed_rate or work_ret != Block::WORK_CALLED_PRODUCE);
-        const size_t items = (enable_fixed_rate)? (myulround((noutput_items/this->relative_rate))) : this->consume_items[i];
+        const size_t items = (this->consume_called[i])? this->consume_items[i] : (myulround((noutput_items/this->relative_rate)));
         this->consume_items[i] = 0;
 
         this->items_consumed[i] += items;
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index 01a9791..11aa305 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -88,6 +88,7 @@ struct ElementImpl
     //track work's calls to produce and consume
     std::vector<size_t> produce_items;
     std::vector<size_t> consume_items;
+    std::vector<bool> consume_called;
 
     //track the subscriber counts
     std::vector<Token> input_tokens;
diff --git a/lib/gras_impl/debug.hpp b/lib/gras_impl/debug.hpp
index c235c1f..e5f334e 100644
--- a/lib/gras_impl/debug.hpp
+++ b/lib/gras_impl/debug.hpp
@@ -24,7 +24,7 @@
 #define ARMAGEDDON 0
 #define MESSAGE 0
 #define WORK 0
-#define ASSERTING 0
+#define ASSERTING 1
 
 #define HERE() std::cerr << __FILE__ << ":" << __LINE__ << std::endl << std::flush;
 #define VAR(x) std::cerr << #x << " = " << (x) << std::endl << std::flush;
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 2188509..7c25e3b 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -124,6 +124,7 @@ struct InputBufferQueues
     std::vector<size_t> _multiple_bytes;
     std::vector<size_t> _post_bytes;
     std::vector<boost::shared_ptr<BufferQueue> > _aux_queues;
+    std::vector<bool> _in_hist_buff;
 };
 
 
@@ -137,6 +138,7 @@ inline void InputBufferQueues::resize(const size_t size)
     _multiple_bytes.resize(size, 0);
     _post_bytes.resize(size, 0);
     _aux_queues.resize(size);
+    _in_hist_buff.resize(size, false);
 }
 
 
@@ -161,6 +163,7 @@ inline void InputBufferQueues::init(
         _multiple_bytes[i] = std::max(size_t(1), _reserve_bytes[i]);
         _post_bytes[i] = input_item_sizes[i]*max_history_items;
         _post_bytes[i] = std::max(_post_bytes[i], _reserve_bytes[i]);
+        _post_bytes[i] += input_item_sizes[i]; //pad for round down issues
 
         //allocate mini buffers for history edge conditions
         size_t num_bytes = _history_bytes[i] + _post_bytes[i];
@@ -179,6 +182,7 @@ inline void InputBufferQueues::init(
             buff.length = 0;
 
             _queues[i].push_front(buff);
+            _in_hist_buff[i] = true;
         }
     }
 }
@@ -233,6 +237,7 @@ inline void InputBufferQueues::__prepare(const size_t i)
             hist_bytes = _history_bytes[i];
             dst.offset = hist_bytes;
             dst.length = 0;
+            _in_hist_buff[i] = true;
         }
 
         SBuffer src = _queues[i].front();
@@ -263,9 +268,6 @@ inline bool InputBufferQueues::consume(const size_t i, const size_t bytes_consum
     //assert that we dont consume past the bounds of the buffer
     ASSERT(_queues[i].front().length >= bytes_consumed);
 
-    //this is an optimization
-    const bool minibuff = (_history_bytes[i] != 0) and (_queues[i].front().offset == _history_bytes[i]) and (bytes_consumed == _post_bytes[i]);
-
     //update bounds on the current buffer
     _queues[i].front().offset += bytes_consumed;
     _queues[i].front().length -= bytes_consumed;
@@ -276,12 +278,23 @@ inline bool InputBufferQueues::consume(const size_t i, const size_t bytes_consum
         _queues[i].pop_front();
     }
 
+    if (_in_hist_buff[i] and _queues[i].front().offset >= 2*_history_bytes[i])
+    {
+        const size_t residual = _queues[i].front().length;
+        _queues[i].pop_front();
+        _in_hist_buff[i] = false;
+        ASSERT(not _queues[i].empty());
+        ASSERT(_queues[i].front().offset > residual);
+        _queues[i].front().offset -= residual;
+        _queues[i].front().length += residual;
+    }
+
     //update the number of bytes in this queue
     ASSERT(_enqueued_bytes[i] >= bytes_consumed);
     _enqueued_bytes[i] -= bytes_consumed;
     __update(i);
 
-    return not minibuff; //not true on minibuff
+    return not _in_hist_buff[i];
 }
 
 } //namespace gnuradio