gras: fixed buffer inlining with explicit support in buffer queue

Rather than push the inlined buffer into the BufferQueue,
the output buffer queues wrapper explicitly supports
holding onto the inlined buffer.

This is because BufferQueue overloads generally do a token check
so they only accept buffers that they are in charge of.

fixes #79

2 files changed, 24 insertions, 6 deletions

diff --git a/lib/gras_impl/output_buffer_queues.hpp b/lib/gras_impl/output_buffer_queues.hpp
index 251de96..f0859a3 100644
--- a/lib/gras_impl/output_buffer_queues.hpp
+++ b/lib/gras_impl/output_buffer_queues.hpp
@@ -31,6 +31,7 @@ struct OutputBufferQueues
         _bitset.resize(size);
         _queues.resize(size);
         _reserve_bytes.resize(size, 1);
+        _inline_buffer.resize(size);
     }
 
     GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buff)
@@ -49,12 +50,19 @@ struct OutputBufferQueues
 
     GRAS_FORCE_INLINE SBuffer &front(const size_t i)
     {
+        if GRAS_UNLIKELY(_inline_buffer[i]) return _inline_buffer[i];
         ASSERT(not this->empty(i));
         return _queues[i]->front();
     }
 
     GRAS_FORCE_INLINE void consume(const size_t i)
     {
+        if GRAS_UNLIKELY(_inline_buffer[i])
+        {
+            _inline_buffer[i].reset();
+            return;
+        }
+
         ASSERT(not this->empty(i));
         SBuffer &buff = this->front(i);
         if GRAS_UNLIKELY(buff.length == 0) return;
@@ -68,6 +76,12 @@ struct OutputBufferQueues
 
     GRAS_FORCE_INLINE void pop(const size_t i)
     {
+        if GRAS_UNLIKELY(_inline_buffer[i])
+        {
+            _inline_buffer[i].reset();
+            return;
+        }
+
         ASSERT(_queues[i]);
         ASSERT(not _queues[i]->empty());
         _queues[i]->pop();
@@ -86,6 +100,7 @@ struct OutputBufferQueues
 
     GRAS_FORCE_INLINE bool empty(const size_t i) const
     {
+        if GRAS_UNLIKELY(_inline_buffer[i]) return false;
         return (not _queues[i] or _queues[i]->empty());
     }
 
@@ -111,9 +126,17 @@ struct OutputBufferQueues
         _bitset.set(i, avail >= _reserve_bytes[i]);
     }
 
+    GRAS_FORCE_INLINE void set_inline(const size_t i, const SBuffer &inline_buffer)
+    {
+        _inline_buffer[i] = inline_buffer;
+        _inline_buffer[i].length = 0;
+        _bitset.set(i);
+    }
+
     BitSet _bitset;
     std::vector<BufferQueueSptr> _queues;
     std::vector<size_t> _reserve_bytes;
+    std::vector<SBuffer> _inline_buffer;
 
 };
 
diff --git a/lib/task_main.cpp b/lib/task_main.cpp
index 6c3c23d..01e38ce 100644
--- a/lib/task_main.cpp
+++ b/lib/task_main.cpp
@@ -49,12 +49,7 @@ void BlockActor::task_main(void)
             output_inline_index < num_outputs and
             buff.get_affinity() == this->buffer_affinity
         ){
-            //copy buffer reference but push with zero length, same offset
-            SBuffer new_obuff = buff;
-            new_obuff.length = 0;
-            this->output_queues.consume(output_inline_index);
-            this->output_queues.push(output_inline_index, new_obuff); //you got inlined!
-            output_inline_index++; //done do this output port again
+            this->output_queues.set_inline(output_inline_index++, buff);
         }
         //*/
     }