gras: moved flush output to consume function in output queues

This cleans up some code in block actor.
The message sending code is now in task_main,
and the consume routine is now the shared one.

2 files changed, 13 insertions, 21 deletions

diff --git a/lib/gras_impl/block_actor.hpp b/lib/gras_impl/block_actor.hpp
index 2799313..5444bb0 100644
--- a/lib/gras_impl/block_actor.hpp
+++ b/lib/gras_impl/block_actor.hpp
@@ -104,7 +104,6 @@ struct BlockActor : Apology::Worker
     void produce(const size_t index, const size_t items);
     void consume(const size_t index, const size_t items);
     void produce_buffer(const size_t index, const SBuffer &buffer);
-    void flush_output(const size_t index);
     void task_kicker(void);
     void update_input_avail(const size_t index);
     bool is_input_done(const size_t index);
@@ -168,26 +167,6 @@ struct BlockActor : Apology::Worker
 
 //-------------- common functions from this BlockActor class ---------//
 
-GRAS_FORCE_INLINE void BlockActor::flush_output(const size_t i)
-{
-    if GRAS_UNLIKELY(this->output_queues.empty(i) or this->output_queues.front(i).length == 0) return;
-    SBuffer &buff = this->output_queues.front(i);
-    if GRAS_LIKELY(this->produce_outputs[i])
-    {
-        this->produce_outputs[i] = false;
-        InputBufferMessage buff_msg;
-        buff_msg.buffer = buff;
-        this->post_downstream(i, buff_msg);
-    }
-
-    //increment buffer for next use
-    buff.offset += buff.length;
-    buff.length = 0;
-
-    //release whatever has been used of the output buffer
-    this->output_queues.pop(i);
-}
-
 GRAS_FORCE_INLINE void BlockActor::task_kicker(void)
 {
     if (this->is_work_allowed()) this->Send(SelfKickMessage(), this->GetAddress());
diff --git a/lib/gras_impl/output_buffer_queues.hpp b/lib/gras_impl/output_buffer_queues.hpp
index deea0e1..251de96 100644
--- a/lib/gras_impl/output_buffer_queues.hpp
+++ b/lib/gras_impl/output_buffer_queues.hpp
@@ -53,6 +53,19 @@ struct OutputBufferQueues
         return _queues[i]->front();
     }
 
+    GRAS_FORCE_INLINE void consume(const size_t i)
+    {
+        ASSERT(not this->empty(i));
+        SBuffer &buff = this->front(i);
+        if GRAS_UNLIKELY(buff.length == 0) return;
+
+        //increment buffer for next use
+        buff.offset += buff.length;
+        buff.length = 0;
+
+        this->pop(i);
+    }
+
     GRAS_FORCE_INLINE void pop(const size_t i)
     {
         ASSERT(_queues[i]);