implement logic for work_noutput_items

6 files changed, 11 insertions, 6 deletions

diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index 07237a8..be53fe6 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -23,7 +23,7 @@ void ElementImpl::handle_block_msg(
     const tsbe::TaskInterface &task_iface,
     const tsbe::Wax &msg
 ){
-    std::cout << "handle_block_msg in " << name << std::endl;
+    //std::cout << "handle_block_msg in " << name << std::endl;
 
     //a buffer has returned from the downstream
     //(all interested consumers have finished with it)
@@ -88,7 +88,7 @@ void ElementImpl::handle_block_msg(
 
 void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface)
 {
-    std::cout << "topology_update in " << name << std::endl;
+    //std::cout << "topology_update in " << name << std::endl;
 
     const size_t num_inputs = task_iface.get_num_inputs();
     const size_t num_outputs = task_iface.get_num_outputs();
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 88ca5cf..59be851 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -89,6 +89,7 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
     //------------------------------------------------------------------
     //-- initialize input buffers before work
     //------------------------------------------------------------------
+    size_t num_input_items = (num_inputs)? ~0 : 0; //so big that it must std::min
     size_t input_tokens_count = 0;
     for (size_t i = 0; i < num_inputs; i++)
     {
@@ -103,12 +104,13 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
         this->input_items[i]._len = items;
         this->work_input_items[i] = info.mem;
         this->work_ninput_items[i] = items;
+        num_input_items = std::min(num_input_items, items);
     }
 
     //------------------------------------------------------------------
     //-- initialize output buffers before work
     //------------------------------------------------------------------
-    size_t num_output_items = ~0; //so big that it must std::min
+    size_t num_output_items = (num_outputs)? ~0 : 0; //so big that it must std::min
     size_t output_tokens_count = 0;
     for (size_t i = 0; i < num_outputs; i++)
     {
@@ -147,6 +149,7 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
     //------------------------------------------------------------------
     //-- the work
     //------------------------------------------------------------------
+    work_noutput_items = (num_inputs)? myulround((num_input_items)*this->relative_rate) : num_output_items;
     const int ret = block_ptr->Work(this->input_items, this->output_items);
     const size_t noutput_items = size_t(ret);
 
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index 5d6d07c..b78165c 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -60,6 +60,7 @@ struct ElementImpl
     std::vector<uint64_t> items_produced;
 
     //work buffers for the classic interface
+    size_t work_noutput_items;
     std::vector<const void *> work_input_items;
     std::vector<void *> work_output_items;
     std::vector<int> work_ninput_items;
diff --git a/lib/gr_block.cpp b/lib/gr_block.cpp
index 61ac6c6..cef94b6 100644
--- a/lib/gr_block.cpp
+++ b/lib/gr_block.cpp
@@ -39,7 +39,7 @@ int gr_block::Work(
     const OutputItems &output_items
 ){
     return this->general_work(
-        (output_items.empty())? input_items[0].size() : output_items[0].size(),
+        (*this)->work_noutput_items,
         (*this)->work_ninput_items,
         (*this)->work_input_items,
         (*this)->work_output_items
diff --git a/lib/gras_impl/input_buffer_queues.hpp b/lib/gras_impl/input_buffer_queues.hpp
index 6ac786f..8e1416a 100644
--- a/lib/gras_impl/input_buffer_queues.hpp
+++ b/lib/gras_impl/input_buffer_queues.hpp
@@ -135,6 +135,7 @@ struct InputBufferQueues
         const tsbe::Buffer &buff = _queues[i].front();
         if (_offset_bytes[i] >= buff.get_length())
         {
+            ASSERT(_offset_bytes[i] == buff.get_length()); //bad to consume more than buffer allows
             if (_history_bytes[i] != 0)
             {
                 char *src_mem = ((char *)buff.get_memory()) + _offset_bytes[i] - _history_bytes[i];
diff --git a/lib/port_handlers.cpp b/lib/port_handlers.cpp
index 2461b7d..36d3e27 100644
--- a/lib/port_handlers.cpp
+++ b/lib/port_handlers.cpp
@@ -23,7 +23,7 @@ void ElementImpl::handle_input_msg(
     const size_t index,
     const tsbe::Wax &msg
 ){
-    std::cout << "handle_input_msg in " << name << std::endl;
+    //std::cout << "handle_input_msg in " << name << std::endl;
 
     //handle incoming stream buffer, push into the queue
     if (msg.type() == typeid(tsbe::Buffer))
@@ -65,7 +65,7 @@ void ElementImpl::handle_output_msg(
     const size_t index,
     const tsbe::Wax &msg
 ){
-    std::cout << "handle_output_msg in " << name << std::endl;
+    //std::cout << "handle_output_msg in " << name << std::endl;
 
     //store the token of the downstream consumer
     if (msg.type() == typeid(Token))