runtime: work on allocator handler

7 files changed, 87 insertions, 8 deletions

diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt
index 9c9c7de..090a244 100644
--- a/lib/CMakeLists.txt
+++ b/lib/CMakeLists.txt
@@ -16,6 +16,7 @@ list(APPEND gnuradio_core_sources
     ${CMAKE_CURRENT_SOURCE_DIR}/element.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_task.cpp
+    ${CMAKE_CURRENT_SOURCE_DIR}/block_allocator.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/block_handlers.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/hier_block.cpp
     ${CMAKE_CURRENT_SOURCE_DIR}/top_block.cpp
diff --git a/lib/block.cpp b/lib/block.cpp
index 0755a2c..5a9b7a1 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -41,6 +41,7 @@ Block::Block(const std::string &name):
     (*this)->block = tsbe::Block(config);
 
     (*this)->block_ptr = this;
+    (*this)->hint = 0;
 }
 
 template <typename V, typename T>
diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp
new file mode 100644
index 0000000..b3f8c4d
--- /dev/null
+++ b/lib/block_allocator.cpp
@@ -0,0 +1,52 @@
+//
+// Copyright 2012 Josh Blum
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Lesser General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU Lesser General Public License for more details.
+//
+// You should have received a copy of the GNU Lesser General Public License
+// along with io_sig program.  If not, see <http://www.gnu.org/licenses/>.
+
+#include "element_impl.hpp"
+#include <boost/bind.hpp>
+
+using namespace gnuradio;
+
+//TODO will need more complicated later
+static void simple_allocator(
+    const tsbe::BufferToken &tok,
+    const size_t num_bytes,
+    const size_t num_buffs
+){
+    for (size_t i = 0; i < num_buffs; i++)
+    {
+        tsbe::BufferConfig config;
+        config.memory = NULL;
+        config.length = num_bytes;
+        config.token = tok;
+        tsbe::Buffer buff(config);
+        //buffer derefs here and the token messages it back to the block
+    }
+}
+
+void ElementImpl::handle_allocation(const tsbe::TaskInterface &task_iface)
+{
+    //allocate output buffers which will also wake up the task
+    const size_t num_outputs = task_iface.get_num_outputs();
+    for (size_t i = 0; i < num_outputs; i++)
+    {
+        size_t items = this->hint;
+        if (items == 0) items = 1024;
+        items = std::max(items, this->output_multiple_items[i]);
+
+        const size_t bytes = items * this->output_items_sizes[i];
+        this->block.set_output_port_allocator(i, boost::bind(&simple_allocator, _1, bytes, 2));
+    }
+}
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index 12acfa3..e5a117f 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -60,8 +60,6 @@ void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface, const t
     fill_item_sizes_from_sig(this->output_items_sizes, this->output_signature, num_outputs);
 
     //resize and fill port properties
-    resize_fill_front(this->input_items_sizes, num_inputs);
-    resize_fill_front(this->output_items_sizes, num_outputs);
     resize_fill_front(this->input_history_items, num_inputs);
     resize_fill_front(this->output_multiple_items, num_outputs);
 
@@ -101,11 +99,12 @@ void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface, const t
     if (state.cast<TopBlockMessage>().what == TopBlockMessage::ACTIVE)
     {
         this->active = true;
-        this->handle_task(task_iface);
+
+        //causes initial processing kick-off for source blocks
+        this->handle_allocation(task_iface);
     }
     if (state.cast<TopBlockMessage>().what == TopBlockMessage::INERT)
     {
         this->active = false;
     }
-
 }
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
index 3a34107..c152fa9 100644
--- a/lib/block_task.cpp
+++ b/lib/block_task.cpp
@@ -22,14 +22,22 @@ using namespace gnuradio;
 
 void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
 {
+    //------------------------------------------------------------------
+    //-- Decide if its possible to continue any processing:
+    //-- Handle task may get called for incoming buffers,
+    //-- however, not all ports may have available buffers.
+    //------------------------------------------------------------------
     const bool all_inputs_ready = (~task_iface.get_inputs_ready()).none();
     const bool all_outputs_ready = (~task_iface.get_outputs_ready()).none();
     if (not (this->active and all_inputs_ready and all_outputs_ready)) return;
 
     const size_t num_inputs = task_iface.get_num_inputs();
     const size_t num_outputs = task_iface.get_num_outputs();
+    const bool is_source = (num_inputs == 0);
 
-    //sort the input tags before working
+    //------------------------------------------------------------------
+    //-- sort the input tags before working
+    //------------------------------------------------------------------
     for (size_t i = 0; i < num_inputs; i++)
     {
         if (not this->input_tags_changed[i]) continue;
@@ -38,6 +46,11 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
         this->input_tags_changed[i] = false;
     }
 
+    //------------------------------------------------------------------
+    //-- Processing time!
+    //------------------------------------------------------------------
+
+    HERE();
 
 
     //0) figure out what we have for input data
@@ -49,10 +62,14 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
     //block_ptr->forecast(100
 
 
+    //TODO set deactive when work returns DONE
 
+    //TODO source blocks should call work again until exhausted output buffers
 
-
-    //trim the input tags that are past the consumption zone
+    //------------------------------------------------------------------
+    //-- trim the input tags that are past the consumption zone
+    //-- and post trimmed tags to the downstream based on policy
+    //------------------------------------------------------------------
     for (size_t i = 0; i < num_inputs; i++)
     {
         std::vector<Tag> &tags_i = this->input_tags[i];
@@ -95,7 +112,9 @@ void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
         if (last != 0) tags_i.erase(tags_i.begin(), tags_i.begin()+last);
     }
 
-    //now commit all tags in the output queue to the downstream msg handler
+    //------------------------------------------------------------------
+    //-- now commit all tags in the output queue to the downstream
+    //------------------------------------------------------------------
     for (size_t i = 0; i < num_outputs; i++)
     {
         BOOST_FOREACH(const Tag &t, this->output_tags[i])
diff --git a/lib/element.cpp b/lib/element.cpp
index 464ad94..e0269e7 100644
--- a/lib/element.cpp
+++ b/lib/element.cpp
@@ -30,6 +30,7 @@ Element::Element(void)
 Element::Element(const std::string &name)
 {
     this->reset(new ElementImpl());
+    VAR(name);
     (*this)->name = name;
     (*this)->unique_id = ++unique_id_pool;
 
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index 3880754..1d3dc6a 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -24,6 +24,10 @@
 #include <gnuradio/block.hpp>
 #include <gr_types.h>
 #include <vector>
+#include <iostream>
+
+#define HERE() std::cerr << __FILE__ << ":" << __LINE__ << std::endl << std::flush;
+#define VAR(x) std::cout << #x << " = " << (x) << std::endl << std::flush;
 
 struct TopBlockMessage
 {
@@ -92,10 +96,12 @@ struct ElementImpl
     }
     //gets the handlers access for forecast and work
     Block *block_ptr;
+    size_t hint; //some kind of allocation hint
 
     //handlers
     void handle_port_msg(const size_t, const tsbe::Wax &);
     void topology_update(const tsbe::TaskInterface &, const tsbe::Wax &);
+    void handle_allocation(const tsbe::TaskInterface &);
     void handle_task(const tsbe::TaskInterface &);
 
     //is the fg running?