summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
m---------Apology0
-rw-r--r--include/gras/block.hpp6
-rw-r--r--include/gras/block_config.hpp7
-rw-r--r--include/gras/element.hpp11
-rw-r--r--include/gras/element.i1
-rw-r--r--include/gras/hier_block.hpp6
-rw-r--r--include/gras/hier_block.i1
-rw-r--r--include/gras/top_block.hpp7
-rw-r--r--include/gras/top_block.i1
-rw-r--r--lib/block.cpp10
-rw-r--r--lib/block_allocator.cpp10
-rw-r--r--lib/block_config.cpp21
-rw-r--r--lib/block_handlers.cpp20
-rw-r--r--lib/element.cpp10
-rw-r--r--lib/element_impl.hpp6
-rw-r--r--lib/gras_impl/block_data.hpp1
-rw-r--r--lib/hier_block.cpp10
-rw-r--r--lib/task_main.cpp2
-rw-r--r--lib/top_block.cpp14
-rw-r--r--lib/top_block_query.cpp12
20 files changed, 91 insertions, 65 deletions
diff --git a/Apology b/Apology
-Subproject ab6fc7615056dd5b6c737e1f4005250fe028103
+Subproject 4b592f3d0c60b43f9dca3f9d7c3b501b7c40d3d
diff --git a/include/gras/block.hpp b/include/gras/block.hpp
index 05809a1..cb38e95 100644
--- a/include/gras/block.hpp
+++ b/include/gras/block.hpp
@@ -38,12 +38,6 @@ struct GRAS_API Block : Element
*/
void set_thread_pool(const ThreadPool &thread_pool);
- //! Get the global block config settings
- const GlobalBlockConfig &global_config(void) const;
-
- //! Get the global block config settings
- GlobalBlockConfig &global_config(void);
-
//! Get the configuration rules of an input port
const InputPortConfig &input_config(const size_t which_input) const;
diff --git a/include/gras/block_config.hpp b/include/gras/block_config.hpp
index 5e5c0cb..c90731b 100644
--- a/include/gras/block_config.hpp
+++ b/include/gras/block_config.hpp
@@ -15,6 +15,13 @@ struct GRAS_API GlobalBlockConfig
GlobalBlockConfig(void);
/*!
+ * Merge the settings from another config.
+ * Non-defaults on this config stay,
+ * defaults will be overwritten.
+ */
+ void merge(const GlobalBlockConfig &config);
+
+ /*!
* Constrain the maximum number of items that
* work can be called with for all output ports.
*
diff --git a/include/gras/element.hpp b/include/gras/element.hpp
index df47bef..ce5d658 100644
--- a/include/gras/element.hpp
+++ b/include/gras/element.hpp
@@ -10,6 +10,7 @@
#include <gras/gras.hpp>
#include <gras/callable.hpp>
+#include <gras/block_config.hpp>
#include <gras/weak_container.hpp>
#include <boost/shared_ptr.hpp>
@@ -66,6 +67,16 @@ struct GRAS_API Element : Callable, boost::shared_ptr<ElementImpl>
std::string to_string(void) const;
/*******************************************************************
+ * config interface
+ ******************************************************************/
+
+ //! Get the global block config settings
+ const GlobalBlockConfig &global_config(void) const;
+
+ //! Get the global block config settings
+ GlobalBlockConfig &global_config(void);
+
+ /*******************************************************************
* identification interface
******************************************************************/
diff --git a/include/gras/element.i b/include/gras/element.i
index d462370..1380a97 100644
--- a/include/gras/element.i
+++ b/include/gras/element.i
@@ -27,6 +27,7 @@ namespace gras
%include <std_string.i>
%import <PMC/PMC.i>
%include <gras/gras.hpp>
+%include <gras/block_config.hpp>
%include <gras/callable.hpp>
%include <gras/element.hpp>
diff --git a/include/gras/hier_block.hpp b/include/gras/hier_block.hpp
index 73eeabd..4a020be 100644
--- a/include/gras/hier_block.hpp
+++ b/include/gras/hier_block.hpp
@@ -16,6 +16,12 @@ struct GRAS_API HierBlock : Element
virtual ~HierBlock(void);
+ /*!
+ * Commit changes to the global configuration.
+ * Must be call to apply changes to the global config.
+ */
+ void commit_config(void);
+
/*******************************************************************
* connection flow interface
******************************************************************/
diff --git a/include/gras/hier_block.i b/include/gras/hier_block.i
index f89252e..4b617c2 100644
--- a/include/gras/hier_block.i
+++ b/include/gras/hier_block.i
@@ -9,6 +9,7 @@
#include <gras/hier_block.hpp>
%}
+%include <gras/gras.hpp>
%import <gras/element.i>
%include <gras/hier_block.hpp>
diff --git a/include/gras/top_block.hpp b/include/gras/top_block.hpp
index 415c03e..3dcf23f 100644
--- a/include/gras/top_block.hpp
+++ b/include/gras/top_block.hpp
@@ -3,7 +3,6 @@
#ifndef INCLUDED_GRAS_TOP_BLOCK_HPP
#define INCLUDED_GRAS_TOP_BLOCK_HPP
-#include <gras/block_config.hpp>
#include <gras/hier_block.hpp>
namespace gras
@@ -17,12 +16,6 @@ struct GRAS_API TopBlock : HierBlock
virtual ~TopBlock(void);
- //! Get the global block config settings
- const GlobalBlockConfig &global_config(void) const;
-
- //! Get the global block config settings
- GlobalBlockConfig &global_config(void);
-
/*!
* Commit changes to the overall flow graph.
* Call this after modifying connections.
diff --git a/include/gras/top_block.i b/include/gras/top_block.i
index 23a4275..07d2482 100644
--- a/include/gras/top_block.i
+++ b/include/gras/top_block.i
@@ -10,7 +10,6 @@
%}
%include <gras/gras.hpp>
-%include <gras/block_config.hpp>
%import <gras/element.i>
%import <gras/hier_block.i>
%include <gras/top_block.hpp>
diff --git a/lib/block.cpp b/lib/block.cpp
index acb05d3..406e33c 100644
--- a/lib/block.cpp
+++ b/lib/block.cpp
@@ -121,16 +121,6 @@ typename V::value_type &vector_get_resize(V &v, const size_t index)
return v[index];
}
-const GlobalBlockConfig &Block::global_config(void) const
-{
- return (*this)->block_data->global_config;
-}
-
-GlobalBlockConfig &Block::global_config(void)
-{
- return (*this)->block_data->global_config;
-}
-
InputPortConfig &Block::input_config(const size_t which_input)
{
return vector_get_resize((*this)->block_data->input_configs, which_input);
diff --git a/lib/block_allocator.cpp b/lib/block_allocator.cpp
index 8246037..6f59ef6 100644
--- a/lib/block_allocator.cpp
+++ b/lib/block_allocator.cpp
@@ -61,11 +61,15 @@ void BlockActor::handle_top_alloc(const TopAllocMessage &, const Theron::Address
const size_t num_outputs = worker->get_num_outputs();
for (size_t i = 0; i < num_outputs; i++)
{
+ size_t reserve_items = data->output_configs[i].reserve_items;
+ size_t maximum_items = data->output_configs[i].maximum_items;
+ if (maximum_items == 0) maximum_items = data->block->global_config().maximum_output_items;
+
const size_t bytes = recommend_length(
data->output_allocation_hints[i],
my_round_up_mult(AT_LEAST_BYTES, data->output_configs[i].item_size),
- data->output_configs[i].reserve_items*data->output_configs[i].item_size,
- data->output_configs[i].maximum_items*data->output_configs[i].item_size
+ reserve_items*data->output_configs[i].item_size,
+ maximum_items*data->output_configs[i].item_size
);
SBufferDeleter deleter = boost::bind(&buffer_returner, this->thread_pool, this->GetAddress(), i, _1);
@@ -74,7 +78,7 @@ void BlockActor::handle_top_alloc(const TopAllocMessage &, const Theron::Address
SBufferConfig config;
config.memory = NULL;
config.length = bytes;
- config.affinity = data->global_config.buffer_affinity;
+ config.affinity = data->block->global_config().buffer_affinity;
config.token = token;
BufferQueueSptr queue = data->block->output_buffer_allocator(i, config);
diff --git a/lib/block_config.cpp b/lib/block_config.cpp
index 165ab47..e34aa0c 100644
--- a/lib/block_config.cpp
+++ b/lib/block_config.cpp
@@ -11,6 +11,27 @@ GlobalBlockConfig::GlobalBlockConfig(void)
interruptible_work = false;
}
+void GlobalBlockConfig::merge(const GlobalBlockConfig &config)
+{
+ //overwrite with global config only if maxium_items is not set (zero)
+ if (this->maximum_output_items == 0)
+ {
+ this->maximum_output_items = config.maximum_output_items;
+ }
+
+ //overwrite with global node affinity setting for buffers if not set
+ if (this->buffer_affinity == -1)
+ {
+ this->buffer_affinity = config.buffer_affinity;
+ }
+
+ //overwrite with global interruptable setting for work if not set
+ if (this->interruptible_work == false)
+ {
+ this->interruptible_work = config.interruptible_work;
+ }
+}
+
InputPortConfig::InputPortConfig(void)
{
item_size = 1;
diff --git a/lib/block_handlers.cpp b/lib/block_handlers.cpp
index ca1ad97..57e24c1 100644
--- a/lib/block_handlers.cpp
+++ b/lib/block_handlers.cpp
@@ -83,29 +83,19 @@ void BlockActor::handle_top_config(
const Theron::Address from
){
MESSAGE_TRACER();
- const GlobalBlockConfig &config = message.config;
+
+ //merge in the non-defaults
+ data->block->global_config().merge(message.config);
//overwrite with global config only if maxium_items is not set (zero)
for (size_t i = 0; i < data->output_configs.size(); i++)
{
if (data->output_configs[i].maximum_items == 0)
{
- data->output_configs[i].maximum_items = config.maximum_output_items;
+ data->output_configs[i].maximum_items = data->block->global_config().maximum_output_items;
}
}
- //overwrite with global node affinity setting for buffers if not set
- if (data->global_config.buffer_affinity == -1)
- {
- data->global_config.buffer_affinity = config.buffer_affinity;
- }
-
- //overwrite with global interruptable setting for work if not set
- if (data->global_config.interruptible_work == false)
- {
- data->global_config.interruptible_work = config.interruptible_work;
- }
-
this->Send(0, from); //ACK
}
@@ -120,7 +110,7 @@ void BlockActor::handle_top_thread_group(
//spawn a new thread if this block is a source
data->thread_group = message.thread_group;
data->interruptible_thread.reset(); //erase old one
- if (data->global_config.interruptible_work)
+ if (data->block->global_config().interruptible_work)
{
data->interruptible_thread = boost::make_shared<InterruptibleThread>(
data->thread_group, boost::bind(&BlockActor::task_work, this)
diff --git a/lib/element.cpp b/lib/element.cpp
index 9f3c5fb..5574c02 100644
--- a/lib/element.cpp
+++ b/lib/element.cpp
@@ -84,6 +84,16 @@ std::string Element::to_string(void) const
return (*this)->repr;
}
+const GlobalBlockConfig &Element::global_config(void) const
+{
+ return (*this)->global_config;
+}
+
+GlobalBlockConfig &Element::global_config(void)
+{
+ return (*this)->global_config;
+}
+
void Element::adopt_element(const std::string &name, const Element &child)
{
if (child->parent) throw std::invalid_argument(str(boost::format(
diff --git a/lib/element_impl.hpp b/lib/element_impl.hpp
index 91497ed..dca7963 100644
--- a/lib/element_impl.hpp
+++ b/lib/element_impl.hpp
@@ -38,7 +38,7 @@ struct ElementImpl
//top block stuff
SharedThreadGroup thread_group;
Token token;
- GlobalBlockConfig top_config;
+ GlobalBlockConfig global_config;
//element tree stuff
Element parent;
@@ -63,14 +63,14 @@ struct ElementImpl
void bcast_prio_msg(const MessageType &msg)
{
Theron::Receiver receiver;
- BOOST_FOREACH(Apology::Worker *w, this->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, this->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
MessageType message = msg;
message.prio_token = actor->prio_token;
actor->GetFramework().Send(message, receiver.GetAddress(), actor->GetAddress());
}
- size_t outstandingCount(this->executor->get_workers().size());
+ size_t outstandingCount(this->topology->get_workers().size());
while (outstandingCount != 0)
{
outstandingCount -= receiver.Wait(outstandingCount);
diff --git a/lib/gras_impl/block_data.hpp b/lib/gras_impl/block_data.hpp
index d6af53d..f4dd9bd 100644
--- a/lib/gras_impl/block_data.hpp
+++ b/lib/gras_impl/block_data.hpp
@@ -69,7 +69,6 @@ struct BlockData
//is the fg running?
BlockState block_state;
- GlobalBlockConfig global_config;
std::vector<std::vector<OutputHintMessage> > output_allocation_hints;
diff --git a/lib/hier_block.cpp b/lib/hier_block.cpp
index abbdec1..5d497de 100644
--- a/lib/hier_block.cpp
+++ b/lib/hier_block.cpp
@@ -3,6 +3,7 @@
#include "element_impl.hpp"
#include <gras/hier_block.hpp>
#include <boost/format.hpp>
+#include <boost/foreach.hpp>
#include <exception>
using namespace gras;
@@ -23,6 +24,15 @@ HierBlock::~HierBlock(void)
//NOP
}
+void HierBlock::commit_config(void)
+{
+ BOOST_FOREACH(Apology::Worker *w, (*this)->topology->get_workers())
+ {
+ BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
+ actor->data->block->global_config().merge((*this)->global_config);
+ }
+}
+
void ElementImpl::hier_block_cleanup(void)
{
this->topology->clear_all();
diff --git a/lib/task_main.cpp b/lib/task_main.cpp
index 5547b08..803a5e3 100644
--- a/lib/task_main.cpp
+++ b/lib/task_main.cpp
@@ -106,7 +106,7 @@ void BlockActor::task_main(void)
buff.unique() and
data->input_configs[i].inline_buffer and
output_inline_index < num_outputs and
- buff.get_affinity() == data->global_config.buffer_affinity
+ buff.get_affinity() == data->block->global_config().buffer_affinity
){
data->output_queues.set_inline(output_inline_index++, buff);
}
diff --git a/lib/top_block.cpp b/lib/top_block.cpp
index 0c8572d..dd2fd5d 100644
--- a/lib/top_block.cpp
+++ b/lib/top_block.cpp
@@ -31,16 +31,6 @@ void ElementImpl::top_block_cleanup(void)
this->executor->commit();
}
-const GlobalBlockConfig &TopBlock::global_config(void) const
-{
- return (*this)->top_config;
-}
-
-GlobalBlockConfig &TopBlock::global_config(void)
-{
- return (*this)->top_config;
-}
-
void TopBlock::commit(void)
{
this->start(); //ok to re-start, means update
@@ -62,7 +52,7 @@ void TopBlock::start(void)
{
//send the global block config before alloc
TopConfigMessage message;
- message.config = (*this)->top_config;
+ message.config = (*this)->global_config;
(*this)->bcast_prio_msg(message);
}
{
@@ -127,7 +117,7 @@ void TopBlock::wait(void)
}
//loop through blocks looking for non-done blocks with done inputs
- BOOST_FOREACH(Apology::Worker *w, (*this)->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, (*this)->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
if (actor->data->block_state == BLOCK_STATE_DONE) has_a_done = true;
diff --git a/lib/top_block_query.cpp b/lib/top_block_query.cpp
index ccc5c29..68a56ef 100644
--- a/lib/top_block_query.cpp
+++ b/lib/top_block_query.cpp
@@ -31,7 +31,7 @@ static ptree query_blocks(ElementImpl *self, const ptree &)
{
ptree root;
ptree e;
- BOOST_FOREACH(Apology::Worker *w, self->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, self->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
ptree prop_e;
@@ -61,7 +61,7 @@ static ptree query_stats(ElementImpl *self, const ptree &query)
//get stats with custom receiver and set high prio
GetStatsReceiver receiver;
size_t outstandingCount(0);
- BOOST_FOREACH(Apology::Worker *w, self->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, self->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
@@ -93,7 +93,7 @@ static ptree query_stats(ElementImpl *self, const ptree &query)
//thread pool counts
std::set<ThreadPool> thread_pools;
- BOOST_FOREACH(Apology::Worker *w, self->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, self->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
thread_pools.insert(actor->thread_pool);
@@ -161,7 +161,7 @@ static ptree query_calls(ElementImpl *self, const ptree &query)
ptree root;
const std::string block_id = query.get<std::string>("block");
const std::string call_name = query.get<std::string>("name");
- BOOST_FOREACH(Apology::Worker *w, self->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, self->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
if (actor->data->block->get_uid() != block_id) continue;
@@ -189,7 +189,7 @@ static std::string query_topology(ElementImpl *self, const ptree &query)
buff += "rankdir=LR;\n";
buff += "node [shape=record, fontsize=10];\n";
- BOOST_FOREACH(Apology::Worker *w, self->executor->get_workers())
+ BOOST_FOREACH(Apology::Worker *w, self->topology->get_workers())
{
BlockActor *actor = dynamic_cast<BlockActor *>(w->get_actor());
std::string in_ports_str, out_ports_str;
@@ -222,7 +222,7 @@ static std::string query_topology(ElementImpl *self, const ptree &query)
);
}
- BOOST_FOREACH(const Apology::Flow &flow, self->executor->get_flat_flows())
+ BOOST_FOREACH(const Apology::Flow &flow, self->topology->get_flat_flows())
{
buff += str(boost::format("%u:out%u -> %u:in%u;\n")
% dynamic_cast<const Apology::Worker *>(flow.src.elem)->get_actor()->GetAddress().AsInteger()
generated by cgit (git 2.34.1) at 2025-03-13 02:13:08 +0000