summaryrefslogtreecommitdiff
path: root/lib
diff options
context:
space:
mode:
Diffstat (limited to 'lib')
-rw-r--r--lib/CMakeLists.txt1
-rw-r--r--lib/block.cpp1
-rw-r--r--lib/block_allocator.cpp52
-rw-r--r--lib/block_handlers.cpp7
-rw-r--r--lib/block_task.cpp27
-rw-r--r--lib/element.cpp1
-rw-r--r--lib/element_impl.hpp6
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?
generated by cgit (git 2.34.1) at 2025-03-13 14:08:12 +0000