various fixes for unit tests pass

5 files changed, 19 insertions, 9 deletions

diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index cc9d6af..efdc574 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -233,7 +233,7 @@ void BlockActor::handle_task(void)
 void BlockActor::consume(const size_t i, const size_t items)
 {
     #ifdef ITEM_CONSPROD
-    std::cerr << "consume " << items << std::endl;
+    std::cerr << name << " consume " << items << std::endl;
     #endif
     this->items_consumed[i] += items;
     const size_t bytes = items*this->input_items_sizes[i];
@@ -244,7 +244,7 @@ void BlockActor::consume(const size_t i, const size_t items)
 void BlockActor::produce(const size_t i, const size_t items)
 {
     #ifdef ITEM_CONSPROD
-    std::cerr << "produce " << items << std::endl;
+    std::cerr << name << " produce " << items << std::endl;
     #endif
     SBuffer &buff = this->output_queues.front(i);
     this->items_produced[i] += items;
diff --git a/lib/buffer_queue_pool.cpp b/lib/buffer_queue_pool.cpp
index a2a4c28..7a531a4 100644
--- a/lib/buffer_queue_pool.cpp
+++ b/lib/buffer_queue_pool.cpp
@@ -8,12 +8,19 @@ using namespace gras;
 
 struct BufferQueuePool : BufferQueue
 {
-    BufferQueuePool(const size_t num):
+    BufferQueuePool(const SBufferConfig &config, const size_t num):
+        token(config.token), //save config, its holds token
         queue(boost::circular_buffer<SBuffer>(num))
     {
         //NOP
     }
 
+    ~BufferQueuePool(void)
+    {
+        token.reset();
+        queue.clear();
+    }
+
     SBuffer &front(void)
     {
         ASSERT(not queue.empty());
@@ -24,13 +31,12 @@ struct BufferQueuePool : BufferQueue
     void pop(void)
     {
         ASSERT(not queue.empty());
-        queue.front() = SBuffer(); //dont hold ref
+        queue.front().reset(); //dont hold ref
         queue.pop_front();
     }
 
     void push(const SBuffer &buff)
     {
-        ASSERT(buff);
         queue.push_back(buff);
     }
 
@@ -39,6 +45,7 @@ struct BufferQueuePool : BufferQueue
         return queue.empty();
     }
 
+    SBufferToken token;
     boost::circular_buffer<SBuffer> queue;
 
 };
@@ -47,12 +54,14 @@ BufferQueueSptr BufferQueue::make_pool(
     const SBufferConfig &config,
     const size_t num_buffs
 ){
-    BufferQueueSptr bq(new BufferQueuePool(num_buffs));
+    BufferQueueSptr queue(new BufferQueuePool(config, num_buffs));
+
     for (size_t i = 0; i < num_buffs; i++)
     {
         SBuffer buff(config);
         std::memset(buff.get_actual_memory(), 0, buff.get_actual_length());
         //bq->push(buff);
     }
-    return bq;
+
+    return queue;
 }
diff --git a/lib/gras_impl/output_buffer_queues.hpp b/lib/gras_impl/output_buffer_queues.hpp
index 27a6175..9ff0807 100644
--- a/lib/gras_impl/output_buffer_queues.hpp
+++ b/lib/gras_impl/output_buffer_queues.hpp
@@ -34,8 +34,7 @@ struct OutputBufferQueues
 
     GRAS_FORCE_INLINE void push(const size_t i, const SBuffer &buff)
     {
-        ASSERT(buff);
-        ASSERT(_queues[i]);
+        if (not _queues[i]) return; //block is likely done, throw out buffer
         _queues[i]->push(buff);
         _update(i);
     }
diff --git a/lib/input_handlers.cpp b/lib/input_handlers.cpp
index 66d2b68..f6a81bc 100644
--- a/lib/input_handlers.cpp
+++ b/lib/input_handlers.cpp
@@ -78,6 +78,7 @@ void BlockActor::handle_input_update(const InputUpdateMessage &message, const Th
     const size_t i = message.index;
 
     //update buffer queue configuration
+    if (i >= this->input_queues.size()) return;
     const size_t preload_bytes = this->input_items_sizes[i]*this->input_configs[i].preload_items;
     const size_t reserve_bytes = this->input_items_sizes[i]*this->input_configs[i].reserve_items;
     const size_t maximum_bytes = this->input_items_sizes[i]*this->input_configs[i].maximum_items;
diff --git a/lib/output_handlers.cpp b/lib/output_handlers.cpp
index 41dd791..aaf5544 100644
--- a/lib/output_handlers.cpp
+++ b/lib/output_handlers.cpp
@@ -79,6 +79,7 @@ void BlockActor::handle_output_update(const OutputUpdateMessage &message, const
     const size_t i = message.index;
 
     //update buffer queue configuration
+    if (i >= this->output_queues.size()) return;
     const size_t reserve_bytes = this->output_items_sizes[i]*this->output_configs[i].reserve_items;
     this->output_queues.set_reserve_bytes(i, reserve_bytes);
 }