fixed input fail logic w/ input_queues.is_front_maximal

lib/block_task.cpp

@@ -59,8 +59,6 @@ void BlockActor::mark_done(void)
 
 void BlockActor::input_fail(const size_t i)
 {
-    SBuffer &buff = this->input_queues.front(i);
-
     //input failed, accumulate and try again
     if (not this->input_queues.is_accumulated(i))
     {
@@ -77,10 +75,9 @@ void BlockActor::input_fail(const size_t i)
     }
 
     //check that the input is not already maxed
-    const size_t front_items = buff.length/this->input_items_sizes[i];
-    if (front_items >= this->input_configs[i].maximum_items)
+    if (this->input_queues.is_front_maximal(i))
     {
-        //throw std::runtime_error("input_fail called on maximum_items buffer");
+        throw std::runtime_error("input_fail called on maximum_items buffer");
     }
 }
 

lib/gras_impl/input_buffer_queues.hpp

@@ -50,11 +50,15 @@ struct InputBufferQueues
      * or the first buffer is larger than we can accumulate.
      */
     GRAS_FORCE_INLINE bool is_accumulated(const size_t i) const
+    {
+        return (_queues[i].size() <= 1) or this->is_front_maximal(i);
+    }
+
+    //! Return true if the front buffer is at least max size
+    GRAS_FORCE_INLINE bool is_front_maximal(const size_t i) const
     {
         ASSERT(not _queues[i].empty());
-        return
-            (_queues[i].front().length == _enqueued_bytes[i]) or
-            (_queues[i].front().length >= _maximum_bytes[i]);
+        return _queues[i].front().length >= _maximum_bytes[i];
     }
 
     GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buffer)
@@ -135,6 +139,7 @@ inline void InputBufferQueues::update_config(
 {
     //first allocate the aux buffer
     if (maximum_bytes != 0) _maximum_bytes[i] = maximum_bytes;
+    _maximum_bytes[i] = std::max(_maximum_bytes[i], reserve_bytes);
     if (
         not _aux_queues[i] or
         _aux_queues[i]->empty() or