remove output bytes offset, can use sbuffer's length

6 files changed, 12 insertions, 24 deletions

diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp
index f69cd54..327a3cd 100644
--- a/lib/block_allocator.cpp
+++ b/lib/block_allocator.cpp
@@ -26,7 +26,7 @@ void ElementImpl::buffer_returner(const size_t index, SBuffer &buffer)
 {
     //reset offset and length
     buffer.offset = 0;
-    buffer.length = buffer.get_actual_length();
+    buffer.length = 0;
 
     BufferReturnMessage message;
     message.index = index;
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index 367a6f3..85520bd 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -141,7 +141,6 @@ void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface)
     this->produce_items.resize(num_outputs, 0);
     this->input_queues.resize(num_inputs);
     this->output_queues.resize(num_outputs);
-    this->output_bytes_offset.resize(num_outputs, 0);
     this->forecast_enable = not this->enable_fixed_rate;
 
     this->input_tokens.resize(num_inputs);
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 4acc1b9..3843b28 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -29,13 +29,11 @@ void ElementImpl::mark_done(const tsbe::TaskInterface &task_iface)
     //flush partial output buffers to the downstream
     for (size_t i = 0; i < task_iface.get_num_outputs(); i++)
     {
-        if (this->output_bytes_offset[i] == 0) continue;
-        ASSERT(this->output_queues.ready(i));
+        if (not this->output_queues.ready(i)) continue;
         SBuffer &buff = this->output_queues.front(i);
-        buff.length = this->output_bytes_offset[i];
+        if (buff.length == 0) continue;
         task_iface.post_downstream(i, buff);
         this->output_queues.pop(i);
-        this->output_bytes_offset[i] = 0;
     }
 
     //mark down the new state
@@ -134,8 +132,8 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
 
         ASSERT(this->output_queues.ready(i));
         const SBuffer &buff = this->output_queues.front(i);
-        void *mem = buff.get(this->output_bytes_offset[i]);
-        const size_t bytes = buff.length - this->output_bytes_offset[i];
+        void *mem = buff.get(buff.length);
+        const size_t bytes = buff.get_actual_length() - buff.length - buff.offset;
         const size_t items = bytes/this->output_items_sizes[i];
 
         this->work_io_ptr_mask |= ptrdiff_t(mem);
@@ -216,20 +214,18 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
         this->produce_items[i] = 0;
         if (items == 0) continue;
 
+        SBuffer &buff = this->output_queues.front(i);
         this->items_produced[i] += items;
         const size_t bytes = items*this->output_items_sizes[i];
-        this->output_bytes_offset[i] += bytes;
+        buff.length += bytes;
 
         //only pass output buffer downstream when the input is fully consumed...
         //Reasoning: For the sake of dealling with history, we can process the mini history input buffer,
         //and then call work again on the real input buffer, but still yield one output buffer per input buffer.
         if (input_allows_flush)
         {
-            SBuffer &buff = this->output_queues.front(i);
-            buff.length = this->output_bytes_offset[i];
             task_iface.post_downstream(i, buff);
             this->output_queues.pop(i);
-            this->output_bytes_offset[i] = 0;
         }
     }
 
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index 0c080fd..2119935 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -85,7 +85,6 @@ struct ElementImpl
     //buffer queues and ready conditions
     InputBufferQueues input_queues;
     VectorOfQueues<SBuffer> output_queues;
-    std::vector<size_t> output_bytes_offset;
 
     //tag tracking
     std::vector<bool> input_tags_changed;
diff --git a/lib/gras_impl/buffer_queue.hpp b/lib/gras_impl/buffer_queue.hpp
index 306b369..b3c9cf7 100644
--- a/lib/gras_impl/buffer_queue.hpp
+++ b/lib/gras_impl/buffer_queue.hpp
@@ -26,16 +26,9 @@ namespace gnuradio
 
 struct BufferQueue : std::queue<SBuffer>
 {
-    void __push(SBuffer &buffer)
-    {
-        buffer.offset = 0;
-        buffer.length = buffer.get_actual_length();
-        this->push(buffer);
-    }
-
     BufferQueue(void)
     {
-        SBufferDeleter deleter = boost::bind(&BufferQueue::__push, this, _1);
+        SBufferDeleter deleter = boost::bind(&BufferQueue::push, this, _1);
         _token = SBufferToken(new SBufferDeleter(deleter));
     }
 
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 3cb5b78..acde25a 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -181,10 +181,11 @@ inline void InputBufferQueues::init(
             _aux_queues[i]->pop();
 
             const size_t hist_bytes = _history_bytes[i];
-            std::memset(buff.get(), 0, hist_bytes);
+            std::memset(buff.get_actual_memory(), 0, hist_bytes);
+            buff.offset = hist_bytes;
+            buff.length = 0;
+
             _queues[i].push_front(buff);
-            _queues[i].front().offset = hist_bytes;
-            _queues[i].front().length = 0;
         }
     }
 }