added input fail logic to conditionally accumulate

3 files changed, 33 insertions, 8 deletions

diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 69e99a3..a291c01 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -73,6 +73,19 @@ void BlockActor::mark_done(void)
         << std::flush;
 }
 
+GRAS_FORCE_INLINE void BlockActor::input_fail(const size_t i)
+{
+    //input failed, accumulate and try again
+    if (not this->input_queues.is_accumulated(i))
+    {
+        this->input_queues.accumulate(i, this->input_items_sizes[i]);
+        this->Push(SelfKickMessage(), Theron::Address());
+        return;
+    }
+    //otherwise check for done, else wait for more
+    if (this->inputs_done[i]) this->mark_done();
+}
+
 void BlockActor::handle_task(void)
 {
     #ifdef WORK_DEBUG
@@ -104,7 +117,7 @@ void BlockActor::handle_task(void)
         this->sort_tags(i);
 
         ASSERT(this->input_queues.ready(i));
-        this->input_queues.accumulate(i, this->input_items_sizes[i]);
+        //this->input_queues.accumulate(i, this->input_items_sizes[i]);
         const SBuffer &buff = this->input_queues.front(i);
         void *mem = buff.get();
         size_t items = buff.length/this->input_items_sizes[i];
@@ -119,8 +132,7 @@ void BlockActor::handle_task(void)
         {
             if (items <= this->input_configs[i].lookahead_items)
             {
-                if (this->inputs_done[i]) this->mark_done();
-                return;
+                this->input_fail(i); return;
             }
             items -= this->input_configs[i].lookahead_items;
         }
@@ -198,12 +210,10 @@ void BlockActor::handle_task(void)
         {
             if (fcast_ninput_items[i] <= work_ninput_items[i]) continue;
 
+            //handle the case of forecast failing
             if (work_noutput_items <= this->output_multiple_items)
             {
-                //handle the case of forecast failing
-                //TODO accumulate input here, only done if inputs done and already accumulated
-                if (this->inputs_done[i]) this->mark_done();
-                return;
+                this->input_fail(i); return;
             }
 
             work_noutput_items = work_noutput_items/2; //backoff regime
diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index 71fd96e..d32c105 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -110,6 +110,7 @@ struct BlockActor : Apology::Worker
     void buffer_returner(const size_t index, SBuffer &buffer);
     void mark_done(void);
     void handle_task(void);
+    void input_fail(const size_t index);
     void sort_tags(const size_t index);
     void trim_tags(const size_t index);
     GRAS_FORCE_INLINE bool any_inputs_done(void)
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 1b663c3..d17c4f0 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -58,6 +58,19 @@ struct InputBufferQueues
 
     void accumulate(const size_t i, const size_t item_size);
 
+    /*!
+     * Can we consider this queue's buffers to be accumulated?
+     * Either the first buffer holds all of the enqueued bytes
+     * or the first buffer is larger than we can accumulate.
+     */
+    GRAS_FORCE_INLINE bool is_accumulated(const size_t i) const
+    {
+        ASSERT(not _queues[i].empty());
+        return
+            (_queues[i].front().length == _enqueued_bytes[i]) or
+            (_queues[i].front().length >= MAX_AUX_BUFF_BYTES);
+    }
+
     GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buffer)
     {
         ASSERT(not _queues[i].full());
@@ -182,7 +195,8 @@ GRAS_FORCE_INLINE void InputBufferQueues::accumulate(const size_t i, const size_
     }
 
     _queues[i].push_front(accum_buff);
-    return;
+
+    ASSERT(this->is_accumulated(i));
 }
 
 GRAS_FORCE_INLINE void InputBufferQueues::consume(const size_t i, const size_t bytes_consumed)