implement buffer stitching on input, output always flushes

3 files changed, 20 insertions, 18 deletions

diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index a0d4ba8..6f70bc2 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -195,19 +195,7 @@ void BlockActor::handle_task(void)
     //------------------------------------------------------------------
     for (size_t i = 0; i < num_outputs; i++)
     {
-        if (not this->output_queues.ready(i)) continue;
-        SBuffer &buff = this->output_queues.front(i);
-        const size_t reserve_bytes = this->output_configs[i].reserve_items/this->output_items_sizes[i];
-
-        //dont always pass output buffers downstream for the sake of efficiency
-        if (
-            not this->input_queues.all_ready() or
-            buff.length*2 > buff.get_actual_length() or
-            (buff.get_actual_length() - buff.length) < reserve_bytes
-        )
-        {
-            this->flush_output(i);
-        }
+        this->flush_output(i);
     }
 
     //------------------------------------------------------------------
@@ -229,7 +217,6 @@ void BlockActor::handle_task(void)
 
 void BlockActor::consume(const size_t i, const size_t items)
 {
-    if (items == 0) return;
     this->items_consumed[i] += items;
     const size_t bytes = items*this->input_items_sizes[i];
     this->input_queues.consume(i, bytes);
@@ -256,10 +243,9 @@ void BlockActor::produce_buffer(const size_t i, const SBuffer &buffer)
 
 GRAS_FORCE_INLINE void BlockActor::flush_output(const size_t i)
 {
-    if (not this->output_queues.ready(i)) return;
-    SBuffer &buff = this->output_queues.front(i);
+    if (not this->output_queues.ready(i) or this->output_queues.front(i).length == 0) return;
     InputBufferMessage buff_msg;
-    buff_msg.buffer = buff;
+    buff_msg.buffer = this->output_queues.front(i);
     this->post_downstream(i, buff_msg);
     this->output_queues.pop(i);
 }
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 5ae8951..1bff51f 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -83,7 +83,20 @@ struct InputBufferQueues
     {
         ASSERT(not _queues[i].full());
         if (buffer.length == 0) return;
-        _queues[i].push_back(buffer);
+
+        //does this buffer starts where the last one ends?
+        //perform buffer stitching into back of buffer
+        if (
+            not _queues[i].empty() and _queues[i].back() == buffer and
+            (_queues[i].back().length + _queues[i].back().offset) == buffer.offset
+        ){
+            _queues[i].back().length += buffer.length;
+        }
+        else
+        {
+            _queues[i].push_back(buffer);
+        }
+
         _enqueued_bytes[i] += _queues[i].back().length;
         __update(i);
     }
@@ -225,6 +238,8 @@ GRAS_FORCE_INLINE void InputBufferQueues::accumulate(const size_t i, const size_
 
 GRAS_FORCE_INLINE void InputBufferQueues::consume(const size_t i, const size_t bytes_consumed)
 {
+    if (bytes_consumed == 0) return;
+
     ASSERT(not _queues[i].empty());
     SBuffer &front = _queues[i].front();
 
diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp
index 74ed8da..0069f5b 100644
--- a/lib/input_handlers.cpp
+++ b/lib/input_handlers.cpp
@@ -13,6 +13,7 @@ void BlockActor::handle_input_tag(const InputTagMessage &message, const Theron::
     //handle incoming stream tag, push into the tag storage
     this->input_tags[index].push_back(message.tag);
     this->input_tags_changed[index] = true;
+    this->handle_task();
 }
 
 void BlockActor::handle_input_buffer(const InputBufferMessage &message, const Theron::Address)