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_handlers.cpp78
-rw-r--r--lib/block_task.cpp107
-rw-r--r--lib/element_impl.hpp17
-rw-r--r--lib/top_block.cpp16
6 files changed, 150 insertions, 70 deletions
diff --git a/lib/CMakeLists.txt b/lib/CMakeLists.txt
index 3b0dc4a..f8d915a 100644
--- a/lib/CMakeLists.txt
+++ b/lib/CMakeLists.txt
@@ -15,6 +15,7 @@ list(APPEND gnuradio_core_sources
${CMAKE_CURRENT_SOURCE_DIR}/gr_msg_handler.cc
${CMAKE_CURRENT_SOURCE_DIR}/element.cpp
${CMAKE_CURRENT_SOURCE_DIR}/block.cpp
+ ${CMAKE_CURRENT_SOURCE_DIR}/block_task.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 d7156f7..0755a2c 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -40,6 +40,7 @@ Block::Block(const std::string &name):
config.task_callback = boost::bind(&ElementImpl::handle_task, this->get(), _1);
(*this)->block = tsbe::Block(config);
+ (*this)->block_ptr = this;
}
template <typename V, typename T>
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index 262abea..12acfa3 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -15,8 +15,6 @@
// along with io_sig program. If not, see <http://www.gnu.org/licenses/>.
#include "element_impl.hpp"
-#include <boost/foreach.hpp>
-#include <algorithm>
using namespace gnuradio;
@@ -25,6 +23,7 @@ void ElementImpl::handle_port_msg(const size_t index, const tsbe::Wax &msg)
if (msg.type() == typeid(Tag))
{
this->input_tags[index].push_back(msg.cast<Tag>());
+ this->input_tags_changed[index] = true;
}
}
@@ -80,72 +79,33 @@ void ElementImpl::topology_update(const tsbe::TaskInterface &task_iface, const t
this->produce_items.resize(num_outputs);
//resize tags vector to match sizes
+ this->input_tags_changed.resize(num_inputs);
this->input_tags.resize(num_inputs);
this->output_tags.resize(num_outputs);
-}
-
-void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
-{
- const size_t num_inputs = task_iface.get_num_inputs();
- const size_t num_outputs = task_iface.get_num_outputs();
-
- //sort the input tags before working
- for (size_t i = 0; i < num_inputs; i++)
- {
- std::vector<Tag> &tags_i = this->input_tags[i];
- std::sort(tags_i.begin(), tags_i.end(), Tag::offset_compare);
- }
- //trim the input tags that are past the consumption zone
+ //resize and clear that initial history
+ this->history_buffs.resize(num_inputs);
for (size_t i = 0; i < num_inputs; i++)
{
- std::vector<Tag> &tags_i = this->input_tags[i];
- const size_t items_consumed_i = this->items_consumed[i];
- size_t last = 0;
- while (last < tags_i.size() and tags_i[last].offset < items_consumed_i)
+ tsbe::Buffer &buff = this->history_buffs[i];
+ const size_t num_bytes = this->input_items_sizes[i]*this->input_history_items[i];
+ if (not buff or buff.get_length() != num_bytes)
{
- last++;
+ tsbe::BufferConfig config;
+ config.memory = NULL;
+ config.length = num_bytes;
+ buff = tsbe::Buffer(config);
}
-
- //follow the tag propagation policy before erasure
- switch (this->tag_prop_policy)
- {
- case Block::TPP_DONT: break; //well that was ez
- case Block::TPP_ALL_TO_ALL:
- for (size_t out_i = 0; out_i < num_outputs; out_i++)
- {
- for (size_t tag_i = 0; tag_i < last; tag_i++)
- {
- Tag t = tags_i[tag_i];
- t.offset *= this->relative_rate;
- task_iface.post_downstream(out_i, t);
- }
- }
- break;
- case Block::TPP_ONE_TO_ONE:
- if (i < num_outputs)
- {
- for (size_t tag_i = 0; tag_i < last; tag_i++)
- {
- Tag t = tags_i[tag_i];
- t.offset *= this->relative_rate;
- task_iface.post_downstream(i, t);
- }
- }
- break;
- };
-
- //now its safe to perform the erasure
- 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
- for (size_t i = 0; i < num_outputs; i++)
+ if (state.cast<TopBlockMessage>().what == TopBlockMessage::ACTIVE)
{
- BOOST_FOREACH(const Tag &t, this->output_tags[i])
- {
- task_iface.post_downstream(i, t);
- }
- this->output_tags[i].clear();
+ this->active = true;
+ this->handle_task(task_iface);
+ }
+ if (state.cast<TopBlockMessage>().what == TopBlockMessage::INERT)
+ {
+ this->active = false;
}
+
}
diff --git a/lib/block_task.cpp b/lib/block_task.cpp
new file mode 100644
index 0000000..3a34107
--- /dev/null
+++ b/lib/block_task.cpp
@@ -0,0 +1,107 @@
+//
+// 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/foreach.hpp>
+#include <algorithm>
+
+using namespace gnuradio;
+
+void ElementImpl::handle_task(const tsbe::TaskInterface &task_iface)
+{
+ 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();
+
+ //sort the input tags before working
+ for (size_t i = 0; i < num_inputs; i++)
+ {
+ if (not this->input_tags_changed[i]) continue;
+ std::vector<Tag> &tags_i = this->input_tags[i];
+ std::sort(tags_i.begin(), tags_i.end(), Tag::offset_compare);
+ this->input_tags_changed[i] = false;
+ }
+
+
+
+ //0) figure out what we have for input data
+ //1) calculate the possible num output items
+ //2) take into account the item multiple
+ //3) allocate some buffers
+ //4) work....
+
+ //block_ptr->forecast(100
+
+
+
+
+
+ //trim the input tags that are past the consumption zone
+ for (size_t i = 0; i < num_inputs; i++)
+ {
+ std::vector<Tag> &tags_i = this->input_tags[i];
+ const size_t items_consumed_i = this->items_consumed[i];
+ size_t last = 0;
+ while (last < tags_i.size() and tags_i[last].offset < items_consumed_i)
+ {
+ last++;
+ }
+
+ //follow the tag propagation policy before erasure
+ switch (this->tag_prop_policy)
+ {
+ case Block::TPP_DONT: break; //well that was ez
+ case Block::TPP_ALL_TO_ALL:
+ for (size_t out_i = 0; out_i < num_outputs; out_i++)
+ {
+ for (size_t tag_i = 0; tag_i < last; tag_i++)
+ {
+ Tag t = tags_i[tag_i];
+ t.offset *= this->relative_rate;
+ task_iface.post_downstream(out_i, t);
+ }
+ }
+ break;
+ case Block::TPP_ONE_TO_ONE:
+ if (i < num_outputs)
+ {
+ for (size_t tag_i = 0; tag_i < last; tag_i++)
+ {
+ Tag t = tags_i[tag_i];
+ t.offset *= this->relative_rate;
+ task_iface.post_downstream(i, t);
+ }
+ }
+ break;
+ };
+
+ //now its safe to perform the erasure
+ 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
+ for (size_t i = 0; i < num_outputs; i++)
+ {
+ BOOST_FOREACH(const Tag &t, this->output_tags[i])
+ {
+ task_iface.post_downstream(i, t);
+ }
+ this->output_tags[i].clear();
+ }
+}
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index f09ddc8..3880754 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -25,7 +25,7 @@
#include <gr_types.h>
#include <vector>
-struct TopBlockUpdateEvent
+struct TopBlockMessage
{
enum
{
@@ -33,7 +33,7 @@ struct TopBlockUpdateEvent
ACTIVE,
INERT,
HINT,
- } state;
+ } what;
size_t hint;
};
@@ -72,12 +72,16 @@ struct ElementImpl
std::vector<size_t> produce_items;
std::vector<size_t> consume_items;
+ //special buffer for dealing with history
+ std::vector<tsbe::Buffer> history_buffs;
+
//tag tracking
+ std::vector<bool> input_tags_changed;
std::vector<std::vector<Tag> > input_tags;
std::vector<std::vector<Tag> > output_tags;
-
Block::tag_propagation_policy_t tag_prop_policy;
+ //topological things
tsbe::Block block;
tsbe::Topology topology;
tsbe::Executor executor;
@@ -86,11 +90,18 @@ struct ElementImpl
if (block) return block;
return topology;
}
+ //gets the handlers access for forecast and work
+ Block *block_ptr;
+ //handlers
void handle_port_msg(const size_t, const tsbe::Wax &);
void topology_update(const tsbe::TaskInterface &, const tsbe::Wax &);
void handle_task(const tsbe::TaskInterface &);
+ //is the fg running?
+ bool active;
+
+ //rate settings
bool enble_fixed_rate;
double relative_rate;
};
diff --git a/lib/top_block.cpp b/lib/top_block.cpp
index 0fade0d..0ae4ce9 100644
--- a/lib/top_block.cpp
+++ b/lib/top_block.cpp
@@ -34,30 +34,30 @@ TopBlock::TopBlock(const std::string &name):
void TopBlock::update(void)
{
- TopBlockUpdateEvent event;
- event.state = TopBlockUpdateEvent::UPDATE;
+ TopBlockMessage event;
+ event.what = TopBlockMessage::UPDATE;
(*this)->executor.update(event);
}
void TopBlock::set_buffer_hint(const size_t hint)
{
- TopBlockUpdateEvent event;
- event.state = TopBlockUpdateEvent::HINT;
+ TopBlockMessage event;
+ event.what = TopBlockMessage::HINT;
event.hint = hint;
(*this)->executor.update(event);
}
void TopBlock::start(void)
{
- TopBlockUpdateEvent event;
- event.state = TopBlockUpdateEvent::ACTIVE;
+ TopBlockMessage event;
+ event.what = TopBlockMessage::ACTIVE;
(*this)->executor.update(event);
}
void TopBlock::stop(void)
{
- TopBlockUpdateEvent event;
- event.state = TopBlockUpdateEvent::INERT;
+ TopBlockMessage event;
+ event.what = TopBlockMessage::INERT;
(*this)->executor.update(event);
}
generated by cgit (git 2.34.1) at 2025-03-13 14:04:19 +0000